houhou/wechat-robot-skills
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: 重构 pdf skill

Browse Source
This commit is contained in:
hp0912 2026-07-25 15:50:37 +08:00
parent 463c4da6fa
commit 06df60d706
9 changed files with 1484 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

222
skills/pdf/SKILL.md
View File

@ -1,111 +1,175 @@
--- ---
name: pdf name: pdf
description: "处理本地 PDF 文件或远程 HTTPS PDF 链接，包括下载、读取、文本与表格提取、页面渲染与视觉审阅、创建、修改、合并、拆分、旋转、表单处理和最终质量校验。当用户提供 .pdf 文件或 HTTPS PDF 地址，或要求生成、编辑、总结、审阅版式重要的 PDF 时使用。" description: "处理本地 PDF 文件或远程 HTTPS PDF 链接，包括安全下载、元数据与页面检查、分段文本和表格提取、页面 PNG 渲染、从文本创建 PDF、合并、拆分、旋转及最终质量校验。当用户提供 .pdf 文件或 HTTPS PDF 地址，或要求总结、读取、生成、编辑、转换或审阅 PDF 时使用。"
--- ---
# PDF 处理 # PDF 处理
## 核心原则 ## 强制执行规则
- 优先进行视觉检查。PDF 的文本提取结果不能代表真实版式，必须把页面渲染为 PNG 后检查。 当前智能体不能执行 shell、任意 Python 代码或系统命令。只能通过 `execute_skill_script` 调用本 Skill 中真实存在的固定脚本。
- 收到远程 HTTPS PDF 链接时，先下载为本地 PDF，再执行任何读取、编辑或渲染操作。
- 创建 PDF 时优先使用 `reportlab`；提取文本或表格时使用 `pdfplumber`；读取元数据、合并、拆分、旋转或处理页面结构时使用 `pypdf`。
- 每次完成影响内容或版式的修改后，重新渲染并检查最新版本。
- 不覆盖用户提供的源文件。编辑时写入新的输出文件。
## 环境约定 - 只调用下表列出的可执行脚本。
- 不执行 `scripts/` 目录，不执行内部模块 `scripts/_pdf_common.py`。
- 不传 `-c`、`python3`、`ls`、`pdftoppm` 或其他 shell/系统命令作为脚本参数。
- 不创建或猜测脚本清单以外的文件。
- 每次检查脚本返回的 JSON；只有 `ok` 为 `true` 时才继续。
- 收到 `ok: false` 时，依据 `error` 调整合法参数或向用户说明失败原因，不要把参数改传给其他脚本碰运气。
环境已预置 Poppler（`pdfinfo`、`pdftoppm`）以及 `reportlab`、`pdfplumber`、`pypdf`。直接使用环境提供的可执行文件和 Python 运行时；不要安装依赖，也不要提示用户安装依赖。如果默认 `python3` 未加载预置模块，先定位并改用环境提供的 Python 运行时。如果 Poppler 报告字体缓存目录不可写，把 `XDG_CACHE_HOME` 临时设置为本次任务目录下的可写缓存目录后重试。 ## 脚本清单
| 脚本 | 用途 | 底层能力 |
| --- | --- | --- |
| `scripts/download_pdf.py` | 下载并校验远程 HTTPS PDF | `urllib`、`pypdf` |
| `scripts/inspect_pdf.py` | 检查页数、加密、元数据、页面尺寸和表单数量 | `pypdf` |
| `scripts/extract_text.py` | 按页、按字符分段提取正文 | `pdfplumber` |
| `scripts/extract_tables.py` | 按页提取表格 | `pdfplumber` |
| `scripts/render_pdf.py` | 把指定页面渲染为 PNG | Poppler `pdftoppm` |
| `scripts/create_pdf.py` | 从 UTF-8 文本或 Markdown 创建 PDF | `reportlab`、`pypdf` |
| `scripts/manage_pdf.py` | 合并、拆分或旋转 PDF | `pypdf` |
| `scripts/cleanup_pdf_temp.py` | 安全删除本次任务临时目录 | Python 文件 API |
环境已预置所有依赖。不要安装依赖，也不要提示用户安装依赖。
## 标准流程 ## 标准流程
1. 获取输入：识别本地 PDF 路径或 HTTPS PDF 链接；远程链接先按“远程 PDF 下载”流程保存到本地。 1. 为任务选择简短目录名，把中间文件放在 `tmp/pdfs/<任务名>/`。
2. 校验文件：确认文件存在、大小大于 0、可被 PDF 工具解析，并用 `pdfinfo` 获取页数、页面尺寸、加密状态等信息。 2. 远程 HTTPS 链接先调用 `download_pdf.py`；本地文件直接进入下一步。
3. 建立工作目录：把中间文件放在 `tmp/pdfs/<任务名>/`，使用稳定且可读的文件名。 3. 调用 `inspect_pdf.py` 检查文件。遇到加密 PDF 且任务需要正文时，向用户索取密码，不要绕过加密。
4. 读取和处理：根据任务使用 `pdfplumber`、`pypdf` 或 `reportlab`，但不要仅依赖文本提取判断页面内容。 4. 阅读或总结时，循环调用 `extract_text.py` 直至 `has_more` 为 `false`；需要表格时再调用 `extract_tables.py`。
5. 渲染页面：使用 Poppler 把待审阅的 PDF 转成 PNG。 5. 调用 `render_pdf.py` 检查首页、复杂页面、无文本页面和用户关心的页面。创建或修改 PDF 时分批渲染全部页面。
6. 视觉检查：逐页检查文字、图片、表格、页眉页脚、页码、分页和留白。 6. 使用提取文本和渲染结果完成回答；明确区分原文内容与基于版式或图表的推断。
7. 输出结果：把最终 PDF 写入 `output/pdf/`，文件名保持稳定且能表达内容。 7. 创建或修改后的最终 PDF 写入 `output/pdf/`，重新执行检查、文本提取和全部页面渲染。
8. 清理中间文件：交付完成后仅删除本次任务对应的 `tmp/pdfs/<任务名>/`，保留最终产物。 8. 最终产物位于临时目录之外且不再需要缓存时，调用 `cleanup_pdf_temp.py` 清理本次任务目录。
## 远程 PDF 下载 ## 下载远程 PDF
只接受 HTTPS 地址。必须使用本 Skill 自带的 `scripts/download_pdf.py` 下载，不要使用 `curl`、shell 脚本或临时编写的下载逻辑。脚本会自动创建目标目录、流式下载、限制重定向协议、控制文件大小，并使用 `pypdf` 校验下载内容。 只接受 HTTPS 地址。完整保留 URL 及查询参数，不在回复、日志摘要或文件名中复述敏感参数。
调用 `scripts/download_pdf.py`：
```text ```text
python3 scripts/download_pdf.py \ --url 'https://example.com/document.pdf' --output 'tmp/pdfs/<任务名>/source.pdf'
--url 'https://example.com/document.pdf' \
--output 'tmp/pdfs/<任务名>/source.pdf'
``` ```
保留完整 URL，包括查询参数，并把它作为一个完整参数传入。目标文件已存在时，脚本默认拒绝覆盖；仅在确认该文件是本次任务生成的下载缓存时使用 `--overwrite`。可选参数： 可选参数：
- `--timeout <秒>`：连接和读取超时，默认 `60`。 - `--timeout <秒>`：默认 `60`。
- `--max-bytes <字节数>`：最大下载大小，默认 `104857600`（100 MiB）。 - `--max-bytes <字节数>`：默认 `104857600`（100 MiB）。
- `--overwrite`：覆盖已存在的目标文件。 - `--overwrite`：仅在目标是本次任务生成的缓存时使用。
脚本成功时退出码为 `0`，并输出： 脚本会创建父目录、流式下载、阻止 HTTPS 重定向降级到 HTTP，并验证 PDF。成功结果包含 `path`、`size_bytes`、`page_count` 和 `encrypted`。
```json ## 检查 PDF
{
"ok": true, 调用 `scripts/inspect_pdf.py`：
"path": "/absolute/path/to/source.pdf",
"size_bytes": 123456, ```text
"page_count": 10, --input 'tmp/pdfs/<任务名>/source.pdf'
"encrypted": false
}
``` ```
脚本失败时退出码为非 `0`，并输出 `{"ok": false, "error": "具体错误"}`。只有收到 `ok: true` 后才能继续读取或渲染该文件。 使用返回的 `page_count`、`encrypted`、`metadata`、`page_layouts` 和 `form_field_count` 判断后续处理方式。不要直接调用 `pdfinfo`。
遵守以下规则： ## 提取正文
- 跟随 HTTPS 重定向，但不允许降级到 HTTP。 首次调用 `scripts/extract_text.py`：
- 不根据 URL 后缀或响应的 `Content-Type` 单独判断文件类型，以实际 PDF 解析结果为准。
- 如果下载结果是 HTML、登录页、错误页或无法解析的内容，停止后续处理并明确说明链接需要授权或真实 PDF 下载地址。
- 如果链接包含签名、令牌或其他敏感查询参数，不在最终回复、日志摘要、错误信息或生成的文件名中复述这些参数；下载脚本也不会在结果中回显 URL。
- 如果 PDF 已加密且任务需要密码，先向用户索取密码；不得尝试绕过加密。
## 读取与审阅 ```text
--input 'tmp/pdfs/<任务名>/source.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 时必须检查全部页面；只读超长文档时可先生成低分辨率预览定位相关页面，再以足够分辨率检查相关页。 默认单次最多处理 8 页、返回 24000 个字符。可使用：
逐页确认： - `--start-page <页码>`、`--end-page <页码>`：页码从 `1` 开始。
- `--start-offset <字符偏移>`：继续读取被字符上限截断的同一页。
- `--max-pages <页数>`、`--max-chars <字符数>`：控制单次输出。
- `--layout`：仅在需要尽量保留版面空格时使用。
- 没有裁切、重叠、溢出、异常空白页或错误分页。 如果结果中 `has_more: true`，继续调用同一脚本，并把返回的 `next_page` 传给 `--start-page`、`next_offset` 传给 `--start-offset`。后续调用必须保留首次调用的 `--end-page`（如果指定）及其他提取选项。重复直到 `has_more: false`。不要只读取第一批文本就总结长文档。
- 没有黑方块、乱码、缺字或不可读的小字号。
- 标题层级、段落间距、页边距和对齐方式一致。
- 表格、图表和图片清晰、完整、标注正确。
- 页眉、页脚、页码和章节衔接正确。
- 引用和参考文献可读，不含工具令牌、占位符或临时路径。
## 交付要求 提取文本为空或很少时，将页面视为可能的扫描件，改用渲染图读取；不要误判为空白 PDF。
- 只有在最新渲染结果不存在可见的内容或格式缺陷后，才交付创建或修改后的 PDF。 ## 提取表格
- 最终文件必须位于 `output/pdf/`，不得把临时 PNG 或下载缓存当作最终产物。
- 回复时给出最终文件路径，并简要说明完成的处理和验证；失败时说明具体原因，不提供依赖安装提示。 调用 `scripts/extract_tables.py`：
```text
--input 'tmp/pdfs/<任务名>/source.pdf' --start-page 1
```
默认单次最多处理 5 页、20 个表格和 2000 个单元格。可用 `--end-page`、`--start-table`、`--max-pages`、`--max-tables`、`--max-cells` 调整。若 `has_more: true`，把 `next_page` 传给 `--start-page`、`next_table` 传给 `--start-table` 后继续，并保留首次调用的 `--end-page`（如果指定）及其他提取选项。
## 渲染页面
调用 `scripts/render_pdf.py`，不要直接执行 `pdftoppm`：
```text
--input 'tmp/pdfs/<任务名>/source.pdf' --output-dir 'tmp/pdfs/<任务名>/rendered' --start-page 1
```
默认 150 DPI、单次最多 10 页。可使用 `--end-page`、`--max-pages`、`--dpi`、`--timeout` 和 `--overwrite`。若 `has_more: true`，使用 `next_page` 继续，并保留首次调用的 `--end-page`（如果指定）、输出目录及其他渲染选项。脚本返回标准化的 `page-0001.png` 文件路径。
对文字较小或图表密集的页面提高 DPI。使用可用的图像查看或识别工具检查返回的 PNG，不要尝试把图片路径交给下载脚本。
## 创建 PDF
先使用 `write_file` 把内容写为 UTF-8 `.txt` 或 `.md` 文件，再调用 `scripts/create_pdf.py`：
```text
--input 'tmp/pdfs/<任务名>/content.md' --output 'output/pdf/<文件名>.pdf' --title '文档标题'
```
脚本支持 Markdown 标题、项目符号和简单表格，自动选择可嵌入的 Unicode 字体并添加页码。可选参数：
- `--page-size <A4|LETTER>`
- `--font-path <TTF或TTC路径>`
- `--font-size <字号>`
- `--margin <points>`
- `--overwrite`
输入内容只使用 ASCII 连字符 `-`；脚本也会把常见 Unicode 横线规范化为 ASCII 连字符。
## 合并、拆分与旋转
调用 `scripts/manage_pdf.py`，第一个参数必须是操作名。
合并：
```text
merge --input 'a.pdf' --input 'b.pdf' --output 'output/pdf/merged.pdf'
```
拆分指定范围：
```text
split --input 'source.pdf' --output-dir 'output/pdf/split' --range 1-3 --range 4-6
```
不传 `--range` 时每页生成一个 PDF。
旋转指定页面：
```text
rotate --input 'source.pdf' --output 'output/pdf/rotated.pdf' --pages '1,3-5' --degrees 90
```
`--degrees` 只能是 `90`、`180` 或 `270`；不传 `--pages` 时旋转全部页面。目标已存在且确认可覆盖时添加 `--overwrite`。
## 清理临时目录
调用 `scripts/cleanup_pdf_temp.py`：
```text
--task-dir 'tmp/pdfs/<任务名>'
```
脚本只允许删除本 Skill 的 `tmp/pdfs/` 下一级任务目录，拒绝删除根目录、仓库目录或其他路径。
## 质量要求
- 不覆盖用户提供的源文件。
- 创建或修改后重新检查页数、页面尺寸、加密状态和文本可读性。
- 逐页确认没有裁切、重叠、溢出、乱码、黑方块、错误分页或异常空白页。
- 检查标题层级、段落间距、页边距、表格、图表、图片、页码及章节衔接。
- 引用和参考文献必须可读，不得残留工具令牌、占位符或临时路径。
- 只有最新渲染结果不存在可见缺陷时才交付创建或修改后的 PDF。

150
skills/pdf/scripts/_pdf_common.py Normal file
View File

@ -0,0 +1,150 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
import os
import tempfile
from pathlib import Path
from typing import Any, Callable, NoReturn, Optional
class SkillArgumentParser(argparse.ArgumentParser):
def error(self, message: str) -> NoReturn:
raise ValueError(f"参数错误：{message}")
def emit(payload: dict[str, Any]) -> None:
print(json.dumps(payload, ensure_ascii=False))
def failure_message(exc: Exception) -> str:
if isinstance(exc, FileExistsError):
return str(exc)
if isinstance(exc, PermissionError):
return "文件处理失败：没有目标路径的访问权限"
if isinstance(exc, OSError):
return f"文件处理失败：{exc}"
return str(exc)
def run_cli(action: Callable[[], dict[str, Any]]) -> int:
try:
result = action()
except Exception as exc:
emit({"ok": False, "error": failure_message(exc)})
return 1
emit({"ok": True, **result})
return 0
def input_pdf(value: str) -> Path:
path = Path(value).expanduser().resolve()
if not path.exists():
raise FileNotFoundError(f"PDF 文件不存在：{path}")
if not path.is_file():
raise ValueError(f"PDF 路径不是文件：{path}")
if path.suffix.lower() != ".pdf":
raise ValueError("PDF 输入路径必须以 .pdf 结尾")
if path.stat().st_size <= 0:
raise ValueError("PDF 文件为空")
return path
def output_pdf(value: str, overwrite: bool) -> Path:
path = Path(value).expanduser().resolve()
if path.suffix.lower() != ".pdf":
raise ValueError("PDF 输出路径必须以 .pdf 结尾")
if path.exists() and not overwrite:
raise FileExistsError(f"目标文件已存在：{path}")
path.parent.mkdir(parents=True, exist_ok=True)
return path
def output_directory(value: str) -> Path:
path = Path(value).expanduser().resolve()
if path.exists() and not path.is_dir():
raise ValueError(f"输出路径不是目录：{path}")
path.mkdir(parents=True, exist_ok=True)
return path
def new_temp_pdf(output: Path) -> Path:
descriptor, name = tempfile.mkstemp(
prefix=f".{output.name}.",
suffix=".part.pdf",
dir=str(output.parent),
)
os.close(descriptor)
return Path(name)
def publish_temp_file(temp_path: Path, output: Path, overwrite: bool) -> None:
if overwrite:
os.replace(temp_path, output)
return
try:
os.link(temp_path, output)
except FileExistsError as exc:
raise FileExistsError(f"目标文件已存在：{output}") from exc
temp_path.unlink()
def parse_page_spec(value: Optional[str], page_count: int) -> list[int]:
if page_count < 1:
return []
if value is None or not value.strip():
return list(range(1, page_count + 1))
pages: set[int] = set()
for raw_part in value.split(","):
part = raw_part.strip()
if not part:
continue
if "-" in part:
pieces = part.split("-", 1)
try:
start = int(pieces[0])
end = int(pieces[1])
except ValueError as exc:
raise ValueError(f"页码范围格式错误：{part}") from exc
if start > end:
raise ValueError(f"页码范围起始值不能大于结束值：{part}")
else:
try:
start = end = int(part)
except ValueError as exc:
raise ValueError(f"页码格式错误：{part}") from exc
if start < 1 or end > page_count:
raise ValueError(f"页码必须在 1 到 {page_count} 之间：{part}")
pages.update(range(start, end + 1))
if not pages:
raise ValueError("页码范围不能为空")
return sorted(pages)
def selected_page_window(
page_count: int,
start_page: int,
end_page: Optional[int],
max_pages: int,
) -> tuple[int, int, Optional[int]]:
if page_count < 1:
raise ValueError("PDF 没有可处理的页面")
if start_page < 1 or start_page > page_count:
raise ValueError(f"start-page 必须在 1 到 {page_count} 之间")
if max_pages < 1:
raise ValueError("max-pages 必须大于 0")
requested_end = page_count if end_page is None else end_page
if requested_end < start_page or requested_end > page_count:
raise ValueError(
f"end-page 必须在 start-page 到 {page_count} 之间"
)
actual_end = min(requested_end, start_page + max_pages - 1)
next_page = actual_end + 1 if actual_end < requested_end else None
return start_page, actual_end, next_page

52
skills/pdf/scripts/cleanup_pdf_temp.py Normal file
View File

@ -0,0 +1,52 @@
#!/usr/bin/env python3
from __future__ import annotations
import shutil
import sys
from pathlib import Path
from typing import Any
from _pdf_common import SkillArgumentParser, run_cli
def _parse_args(argv: list[str]):
parser = SkillArgumentParser(description="清理当前 PDF 任务的临时目录")
parser.add_argument(
"--task-dir",
required=True,
help="仅允许删除本 Skill 的 tmp/pdfs/ 下某个具体任务目录",
)
return parser.parse_args(argv)
def _cleanup(value: str) -> dict[str, Any]:
skill_root = Path(__file__).resolve().parents[1]
allowed_root = (skill_root / "tmp" / "pdfs").resolve()
candidate = Path(value).expanduser()
target = (
candidate.resolve()
if candidate.is_absolute()
else (skill_root / candidate).resolve()
)
try:
relative = target.relative_to(allowed_root)
except ValueError as exc:
raise ValueError(f"只能清理 {allowed_root} 下的任务目录") from exc
if not relative.parts:
raise ValueError("不能删除 tmp/pdfs 根目录")
if len(relative.parts) != 1:
raise ValueError("task-dir 必须直接指向 tmp/pdfs 下的单个任务目录")
if not target.is_dir():
raise FileNotFoundError(f"任务临时目录不存在：{target}")
shutil.rmtree(target)
return {"removed": str(target)}
def main(argv: list[str] | None = None) -> int:
arguments = sys.argv[1:] if argv is None else argv
return run_cli(lambda: _cleanup(_parse_args(arguments).task_dir))
if __name__ == "__main__":
raise SystemExit(main())

368
skills/pdf/scripts/create_pdf.py Normal file
View File

@ -0,0 +1,368 @@
#!/usr/bin/env python3
from __future__ import annotations
import os
import re
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Any, Optional
from xml.sax.saxutils import escape
from _pdf_common import (
SkillArgumentParser,
new_temp_pdf,
output_pdf,
publish_temp_file,
run_cli,
)
FONT_NAME = "PDFSkillDocumentFont"
ASCII_DASHES = str.maketrans(
{
"\u2010": "-",
"\u2011": "-",
"\u2012": "-",
"\u2013": "-",
"\u2014": "-",
"\u2212": "-",
}
)
def _parse_args(argv: list[str]):
parser = SkillArgumentParser(description="从 UTF-8 文本创建排版后的 PDF")
parser.add_argument("--input", required=True, help="UTF-8 文本或 Markdown 文件")
parser.add_argument("--output", required=True, help="PDF 输出路径")
parser.add_argument("--title", default="", help="文档标题")
parser.add_argument(
"--page-size",
choices=("A4", "LETTER"),
default="A4",
help="页面大小，默认 A4",
)
parser.add_argument(
"--font-path",
default="",
help="可选的 TTF/TTC 字体路径；包含中文时会自动查找预置字体",
)
parser.add_argument(
"--font-size",
type=float,
default=11,
help="正文字号，默认 11",
)
parser.add_argument(
"--margin",
type=float,
default=54,
help="页边距（points），默认 54",
)
parser.add_argument("--overwrite", action="store_true", help="覆盖已有 PDF")
args = parser.parse_args(argv)
if args.font_size < 6 or args.font_size > 36:
raise ValueError("font-size 必须在 6 到 36 之间")
if args.margin < 18 or args.margin > 144:
raise ValueError("margin 必须在 18 到 144 之间")
return args
def _input_text(value: str) -> tuple[Path, str]:
path = Path(value).expanduser().resolve()
if not path.is_file():
raise FileNotFoundError(f"文本输入文件不存在：{path}")
text = path.read_text(encoding="utf-8").translate(ASCII_DASHES)
if not text.strip():
raise ValueError("文本输入文件为空")
return path, text
def _font_candidates(explicit: str, text: str) -> list[Path]:
candidates: list[Path] = []
if explicit:
candidates.append(Path(explicit).expanduser().resolve())
environment_font = os.environ.get("PDF_FONT_PATH", "").strip()
if environment_font:
candidates.append(Path(environment_font).expanduser().resolve())
has_cjk = bool(re.search(r"[\u2e80-\u9fff\uf900-\ufaff]", text))
if has_cjk:
names = [
"/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
"/usr/share/fonts/opentype/noto/NotoSansCJKsc-Regular.otf",
"/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc",
"/System/Library/Fonts/STHeiti Medium.ttc",
"/System/Library/Fonts/Hiragino Sans GB.ttc",
"/Library/Fonts/Arial Unicode.ttf",
]
families = ("Noto Sans CJK SC", "Source Han Sans SC", "Hiragino Sans GB")
else:
names = [
"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
"/System/Library/Fonts/Supplemental/Arial.ttf",
]
families = ("DejaVu Sans", "Arial")
candidates.extend(Path(name) for name in names)
fc_match = shutil_which("fc-match")
if fc_match:
with tempfile.TemporaryDirectory(prefix="pdf-font-cache-") as cache_dir:
environment = os.environ.copy()
environment["XDG_CACHE_HOME"] = cache_dir
for family in families:
try:
completed = subprocess.run(
[fc_match, "-f", "%{file}\n", family],
env=environment,
text=True,
capture_output=True,
timeout=15,
check=False,
)
except (OSError, subprocess.TimeoutExpired):
continue
for line in completed.stdout.splitlines():
if line.strip():
candidates.append(Path(line.strip()))
unique: list[Path] = []
seen: set[str] = set()
for candidate in candidates:
key = str(candidate)
if key not in seen:
seen.add(key)
unique.append(candidate)
return unique
def shutil_which(command: str) -> Optional[str]:
import shutil
return shutil.which(command)
def _register_font(text: str, explicit: str) -> tuple[str, Optional[str]]:
if all(ord(character) < 128 for character in text):
return "Helvetica", None
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont
errors = []
for candidate in _font_candidates(explicit, text):
if not candidate.is_file():
continue
try:
pdfmetrics.registerFont(TTFont(FONT_NAME, str(candidate)))
return FONT_NAME, str(candidate)
except Exception as exc:
errors.append(f"{candidate.name}: {exc}")
detail = errors[-1] if errors else "未找到候选字体"
raise RuntimeError(f"环境预置字体无法覆盖文档字符：{detail}")
def _markdown_table(lines: list[str]) -> Optional[list[list[str]]]:
if len(lines) < 2 or not all("|" in line for line in lines):
return None
rows = [
[cell.strip() for cell in line.strip().strip("|").split("|")]
for line in lines
]
if not rows or not rows[0]:
return None
separator = rows[1]
if len(separator) != len(rows[0]) or not all(
re.fullmatch(r":?-{3,}:?", cell) for cell in separator
):
return None
data = [rows[0], *rows[2:]]
if any(len(row) != len(rows[0]) for row in data):
return None
return data
def _story(
text: str,
title: str,
font_name: str,
font_size: float,
available_width: float,
):
from reportlab.lib import colors
from reportlab.lib.enums import TA_CENTER
from reportlab.lib.styles import ParagraphStyle
from reportlab.lib.units import mm
from reportlab.platypus import Paragraph, Spacer, Table, TableStyle
body = ParagraphStyle(
"Body",
fontName=font_name,
fontSize=font_size,
leading=font_size * 1.5,
spaceAfter=font_size * 0.7,
wordWrap="CJK",
)
heading1 = ParagraphStyle(
"Heading1",
parent=body,
fontSize=font_size * 1.7,
leading=font_size * 2,
spaceBefore=font_size,
spaceAfter=font_size,
)
heading2 = ParagraphStyle(
"Heading2",
parent=body,
fontSize=font_size * 1.35,
leading=font_size * 1.7,
spaceBefore=font_size * 0.8,
spaceAfter=font_size * 0.6,
)
heading3 = ParagraphStyle(
"Heading3",
parent=body,
fontSize=font_size * 1.15,
leading=font_size * 1.5,
spaceBefore=font_size * 0.6,
spaceAfter=font_size * 0.4,
)
title_style = ParagraphStyle(
"Title",
parent=heading1,
fontSize=font_size * 2,
leading=font_size * 2.4,
alignment=TA_CENTER,
spaceAfter=10 * mm,
)
story = []
if title.strip():
story.append(Paragraph(escape(title.strip()), title_style))
for block in re.split(r"\n\s*\n", text.strip()):
lines = [line.rstrip() for line in block.splitlines()]
if not lines:
continue
heading = re.fullmatch(r"(#{1,3})\s+(.+)", lines[0].strip())
table_data = _markdown_table(lines)
if table_data is not None:
cell_style = ParagraphStyle(
"TableCell",
parent=body,
fontSize=max(7, font_size * 0.85),
leading=max(9, font_size * 1.15),
spaceAfter=0,
)
cells = [
[Paragraph(escape(cell), cell_style) for cell in row]
for row in table_data
]
column_width = available_width / len(cells[0])
table = Table(
cells,
colWidths=[column_width] * len(cells[0]),
repeatRows=1,
hAlign="LEFT",
)
table.setStyle(
TableStyle(
[
("FONTNAME", (0, 0), (-1, -1), font_name),
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#E9EEF5")),
("GRID", (0, 0), (-1, -1), 0.5, colors.HexColor("#8793A1")),
("VALIGN", (0, 0), (-1, -1), "TOP"),
("LEFTPADDING", (0, 0), (-1, -1), 6),
("RIGHTPADDING", (0, 0), (-1, -1), 6),
("TOPPADDING", (0, 0), (-1, -1), 5),
("BOTTOMPADDING", (0, 0), (-1, -1), 5),
]
)
)
story.append(table)
elif heading and len(lines) == 1:
styles = {"#": heading1, "##": heading2, "###": heading3}
story.append(Paragraph(escape(heading.group(2)), styles[heading.group(1)]))
elif all(line.lstrip().startswith("- ") for line in lines if line.strip()):
for line in lines:
if line.strip():
story.append(
Paragraph(
escape(line.lstrip()[2:]),
body,
bulletText="-",
)
)
else:
content = "<br/>".join(escape(line) for line in lines)
story.append(Paragraph(content, body))
story.append(Spacer(1, font_size * 0.35))
return story
def _create(args) -> dict[str, Any]:
from pypdf import PdfReader
from reportlab.lib.pagesizes import A4, LETTER
from reportlab.platypus import SimpleDocTemplate
input_path, text = _input_text(args.input)
title = args.title.strip().translate(ASCII_DASHES)
font_name, font_path = _register_font(f"{title}\n{text}", args.font_path)
output = output_pdf(args.output, args.overwrite)
temporary = new_temp_pdf(output)
page_size = A4 if args.page_size == "A4" else LETTER
def draw_footer(canvas, document) -> None:
canvas.saveState()
canvas.setTitle(title or input_path.stem)
canvas.setFont(font_name, 8)
canvas.drawCentredString(page_size[0] / 2, 18, str(document.page))
canvas.restoreState()
try:
document = SimpleDocTemplate(
str(temporary),
pagesize=page_size,
leftMargin=args.margin,
rightMargin=args.margin,
topMargin=args.margin,
bottomMargin=max(args.margin, 36),
title=title or input_path.stem,
)
document.build(
_story(
text,
title,
font_name,
args.font_size,
page_size[0] - (2 * args.margin),
),
onFirstPage=draw_footer,
onLaterPages=draw_footer,
)
with temporary.open("rb") as stream:
reader = PdfReader(stream, strict=False)
page_count = len(reader.pages)
publish_temp_file(temporary, output, args.overwrite)
finally:
temporary.unlink(missing_ok=True)
return {
"input": str(input_path),
"output": str(output),
"page_count": page_count,
"page_size": args.page_size,
"font": font_name,
"font_path": font_path,
}
def main(argv: list[str] | None = None) -> int:
arguments = sys.argv[1:] if argv is None else argv
return run_cli(lambda: _create(_parse_args(arguments)))
if __name__ == "__main__":
raise SystemExit(main())

142
skills/pdf/scripts/extract_tables.py Normal file
View File

@ -0,0 +1,142 @@
#!/usr/bin/env python3
from __future__ import annotations
import sys
from typing import Any
from _pdf_common import (
SkillArgumentParser,
input_pdf,
run_cli,
selected_page_window,
)
DEFAULT_MAX_PAGES = 5
DEFAULT_MAX_TABLES = 20
DEFAULT_MAX_CELLS = 2000
def _parse_args(argv: list[str]):
parser = SkillArgumentParser(description="按页提取 PDF 表格")
parser.add_argument("--input", required=True, help="本地 PDF 路径")
parser.add_argument("--start-page", type=int, default=1, help="起始页，1-based")
parser.add_argument("--end-page", type=int, help="结束页，1-based，默认到末页")
parser.add_argument(
"--start-table",
type=int,
default=0,
help="起始页内从第几个表格继续，0-based",
)
parser.add_argument(
"--max-pages",
type=int,
default=DEFAULT_MAX_PAGES,
help=f"单次最多处理页数，默认 {DEFAULT_MAX_PAGES}",
)
parser.add_argument(
"--max-tables",
type=int,
default=DEFAULT_MAX_TABLES,
help=f"单次最多返回表格数，默认 {DEFAULT_MAX_TABLES}",
)
parser.add_argument(
"--max-cells",
type=int,
default=DEFAULT_MAX_CELLS,
help=f"单次最多返回单元格数，默认 {DEFAULT_MAX_CELLS}",
)
args = parser.parse_args(argv)
if args.start_table < 0:
raise ValueError("start-table 不能小于 0")
if args.max_tables < 1:
raise ValueError("max-tables 必须大于 0")
if args.max_cells < 1:
raise ValueError("max-cells 必须大于 0")
return args
def _clean_table(table) -> list[list[str]]:
return [
["" if cell is None else str(cell).replace("\x00", "") for cell in row]
for row in table
]
def _extract(args) -> dict[str, Any]:
import pdfplumber
path = input_pdf(args.input)
result_pages: list[dict[str, Any]] = []
table_count = 0
cell_count = 0
truncated = False
next_table = 0
with pdfplumber.open(path) as pdf:
page_count = len(pdf.pages)
start_page, actual_end, next_page = selected_page_window(
page_count,
args.start_page,
args.end_page,
args.max_pages,
)
for page_number in range(start_page, actual_end + 1):
page_tables = []
all_tables = pdf.pages[page_number - 1].extract_tables() or []
table_offset = args.start_table if page_number == start_page else 0
if table_offset > len(all_tables):
raise ValueError(
f"start-table 超过第 {page_number} 页表格数量 {len(all_tables)}"
)
for table_index, table in enumerate(
all_tables[table_offset:],
start=table_offset,
):
cleaned = _clean_table(table)
cells = sum(len(row) for row in cleaned)
if cells > args.max_cells and table_count == 0:
raise ValueError(
f"第 {page_number} 页第 {table_index} 个表格有 {cells} "
"个单元格，请提高 max-cells"
)
if (
table_count >= args.max_tables
or cell_count + cells > args.max_cells
):
truncated = True
next_page = page_number
next_table = table_index
break
page_tables.append(cleaned)
table_count += 1
cell_count += cells
result_pages.append({"page": page_number, "tables": page_tables})
if truncated:
break
return {
"path": str(path),
"page_count": page_count,
"start_page": start_page,
"end_page": result_pages[-1]["page"],
"window_end_page": actual_end,
"table_count": table_count,
"cell_count": cell_count,
"truncated": truncated,
"pages": result_pages,
"has_more": next_page is not None,
"next_page": next_page,
"next_table": next_table,
}
def main(argv: list[str] | None = None) -> int:
arguments = sys.argv[1:] if argv is None else argv
return run_cli(lambda: _extract(_parse_args(arguments)))
if __name__ == "__main__":
raise SystemExit(main())

145
skills/pdf/scripts/extract_text.py Normal file
View File

@ -0,0 +1,145 @@
#!/usr/bin/env python3
from __future__ import annotations
import sys
from typing import Any, Optional
from _pdf_common import (
SkillArgumentParser,
input_pdf,
run_cli,
selected_page_window,
)
DEFAULT_MAX_PAGES = 8
DEFAULT_MAX_CHARS = 24000
def _parse_args(argv: list[str]):
parser = SkillArgumentParser(description="按页分段提取 PDF 文本")
parser.add_argument("--input", required=True, help="本地 PDF 路径")
parser.add_argument("--start-page", type=int, default=1, help="起始页，1-based")
parser.add_argument("--end-page", type=int, help="结束页，1-based，默认到末页")
parser.add_argument(
"--start-offset",
type=int,
default=0,
help="起始页文本字符偏移量，用于继续读取被截断的单页",
)
parser.add_argument(
"--max-pages",
type=int,
default=DEFAULT_MAX_PAGES,
help=f"单次最多处理页数，默认 {DEFAULT_MAX_PAGES}",
)
parser.add_argument(
"--max-chars",
type=int,
default=DEFAULT_MAX_CHARS,
help=f"单次最多返回文本字符数，默认 {DEFAULT_MAX_CHARS}",
)
parser.add_argument(
"--layout",
action="store_true",
help="尽量保留版面空格；普通总结通常不要启用",
)
args = parser.parse_args(argv)
if args.start_offset < 0:
raise ValueError("start-offset 不能小于 0")
if args.max_chars < 1:
raise ValueError("max-chars 必须大于 0")
return args
def _extract(args) -> dict[str, Any]:
import pdfplumber
path = input_pdf(args.input)
pages_output: list[dict[str, Any]] = []
used_chars = 0
next_page: Optional[int] = None
next_offset = 0
with pdfplumber.open(path) as pdf:
page_count = len(pdf.pages)
start_page, actual_end, window_next = selected_page_window(
page_count,
args.start_page,
args.end_page,
args.max_pages,
)
for page_number in range(start_page, actual_end + 1):
page = pdf.pages[page_number - 1]
text = page.extract_text(layout=args.layout) or ""
text = text.replace("\x00", "")
offset = args.start_offset if page_number == start_page else 0
if offset > len(text):
raise ValueError(
f"start-offset 超过第 {page_number} 页文本长度 {len(text)}"
)
remaining = text[offset:]
budget = args.max_chars - used_chars
if budget <= 0:
next_page = page_number
next_offset = offset
break
if len(remaining) > budget:
if pages_output and budget < min(500, len(remaining)):
next_page = page_number
next_offset = offset
break
excerpt = remaining[:budget]
pages_output.append(
{
"page": page_number,
"text": excerpt,
"char_count": len(text),
"offset_start": offset,
"offset_end": offset + len(excerpt),
"complete": False,
}
)
used_chars += len(excerpt)
next_page = page_number
next_offset = offset + len(excerpt)
break
pages_output.append(
{
"page": page_number,
"text": remaining,
"char_count": len(text),
"offset_start": offset,
"offset_end": len(text),
"complete": True,
}
)
used_chars += len(remaining)
if next_page is None:
next_page = window_next
return {
"path": str(path),
"page_count": page_count,
"start_page": start_page,
"end_page": pages_output[-1]["page"] if pages_output else None,
"window_end_page": actual_end,
"returned_chars": used_chars,
"pages": pages_output,
"has_more": next_page is not None,
"next_page": next_page,
"next_offset": next_offset,
}
def main(argv: list[str] | None = None) -> int:
arguments = sys.argv[1:] if argv is None else argv
return run_cli(lambda: _extract(_parse_args(arguments)))
if __name__ == "__main__":
raise SystemExit(main())

80
skills/pdf/scripts/inspect_pdf.py Normal file
View File

@ -0,0 +1,80 @@
#!/usr/bin/env python3
from __future__ import annotations
import sys
from collections import Counter
from typing import Any
from _pdf_common import SkillArgumentParser, input_pdf, run_cli
def _parse_args(argv: list[str]):
parser = SkillArgumentParser(description="检查 PDF 元数据和页面结构")
parser.add_argument("--input", required=True, help="本地 PDF 路径")
return parser.parse_args(argv)
def _inspect(path) -> dict[str, Any]:
from pypdf import PdfReader
with path.open("rb") as stream:
reader = PdfReader(stream, strict=False)
encrypted = bool(reader.is_encrypted)
metadata = {
str(key).lstrip("/"): str(value)
for key, value in (reader.metadata or {}).items()
if value is not None
}
if encrypted:
return {
"path": str(path),
"size_bytes": path.stat().st_size,
"encrypted": True,
"page_count": None,
"metadata": metadata,
"page_layouts": [],
"form_field_count": None,
}
layouts: Counter[tuple[float, float, int]] = Counter()
for page in reader.pages:
width = round(float(page.mediabox.width), 2)
height = round(float(page.mediabox.height), 2)
rotation = int(page.get("/Rotate", 0) or 0) % 360
layouts[(width, height, rotation)] += 1
try:
fields = reader.get_fields() or {}
form_field_count = len(fields)
except Exception:
form_field_count = 0
return {
"path": str(path),
"size_bytes": path.stat().st_size,
"encrypted": False,
"page_count": len(reader.pages),
"metadata": metadata,
"page_layouts": [
{
"width_points": width,
"height_points": height,
"rotation": rotation,
"count": count,
}
for (width, height, rotation), count in sorted(layouts.items())
],
"form_field_count": form_field_count,
}
def main(argv: list[str] | None = None) -> int:
arguments = sys.argv[1:] if argv is None else argv
return run_cli(
lambda: _inspect(input_pdf(_parse_args(arguments).input))
)
if __name__ == "__main__":
raise SystemExit(main())

213
skills/pdf/scripts/manage_pdf.py Normal file
View File

@ -0,0 +1,213 @@
#!/usr/bin/env python3
from __future__ import annotations
import re
import sys
from contextlib import ExitStack
from pathlib import Path
from typing import Any
from _pdf_common import (
SkillArgumentParser,
input_pdf,
new_temp_pdf,
output_directory,
output_pdf,
parse_page_spec,
publish_temp_file,
run_cli,
)
def _parse_args(argv: list[str]):
parser = SkillArgumentParser(description="合并、拆分或旋转 PDF 页面")
operations = parser.add_subparsers(dest="operation", required=True)
merge = operations.add_parser("merge", help="合并多个 PDF")
merge.add_argument("--input", action="append", required=True, help="输入 PDF，可重复")
merge.add_argument("--output", required=True, help="合并后的 PDF")
merge.add_argument("--overwrite", action="store_true")
split = operations.add_parser("split", help="按页码范围拆分 PDF")
split.add_argument("--input", required=True)
split.add_argument("--output-dir", required=True)
split.add_argument(
"--range",
dest="ranges",
action="append",
help="页码或页码范围，如 1-3；可重复。缺省时每页一个文件",
)
split.add_argument("--overwrite", action="store_true")
rotate = operations.add_parser("rotate", help="旋转指定页面")
rotate.add_argument("--input", required=True)
rotate.add_argument("--output", required=True)
rotate.add_argument("--pages", help="页码列表，如 1,3-5；缺省时旋转全部页面")
rotate.add_argument("--degrees", type=int, choices=(90, 180, 270), required=True)
rotate.add_argument("--overwrite", action="store_true")
return parser.parse_args(argv)
def _metadata(reader) -> dict[str, str]:
return {
str(key): str(value)
for key, value in (reader.metadata or {}).items()
if value is not None
}
def _merge(args) -> dict[str, Any]:
from pypdf import PdfReader, PdfWriter
inputs = [input_pdf(value) for value in args.input]
if len(inputs) < 2:
raise ValueError("merge 至少需要两个 --input")
output = output_pdf(args.output, args.overwrite)
if output in inputs:
raise ValueError("输出文件不能与输入文件相同")
temporary = new_temp_pdf(output)
writer = PdfWriter()
try:
with ExitStack() as stack:
readers = []
for path in inputs:
stream = stack.enter_context(path.open("rb"))
reader = PdfReader(stream, strict=False)
if reader.is_encrypted:
raise ValueError(f"PDF 已加密，无法合并：{path}")
readers.append(reader)
for page in reader.pages:
writer.add_page(page)
if readers:
writer.add_metadata(_metadata(readers[0]))
with temporary.open("wb") as stream:
writer.write(stream)
page_count = len(writer.pages)
publish_temp_file(temporary, output, args.overwrite)
finally:
writer.close()
temporary.unlink(missing_ok=True)
return {
"operation": "merge",
"inputs": [str(path) for path in inputs],
"output": str(output),
"page_count": page_count,
}
def _split(args) -> dict[str, Any]:
from pypdf import PdfReader, PdfWriter
path = input_pdf(args.input)
destination = output_directory(args.output_dir)
with path.open("rb") as stream:
reader = PdfReader(stream, strict=False)
if reader.is_encrypted:
raise ValueError("PDF 已加密，无法拆分")
page_count = len(reader.pages)
if args.ranges:
groups = [
(value, parse_page_spec(value, page_count))
for value in args.ranges
]
else:
groups = [(str(page), [page]) for page in range(1, page_count + 1)]
outputs = []
for label, pages in groups:
safe_label = re.sub(r"[^0-9_-]+", "_", label.replace(",", "_"))
output = destination / f"{path.stem}_pages_{safe_label}.pdf"
if output.exists() and not args.overwrite:
raise FileExistsError(f"目标文件已存在：{output}")
outputs.append((output, pages))
created = []
for output, pages in outputs:
temporary = new_temp_pdf(output)
writer = PdfWriter()
try:
for page_number in pages:
writer.add_page(reader.pages[page_number - 1])
writer.add_metadata(_metadata(reader))
with temporary.open("wb") as output_stream:
writer.write(output_stream)
publish_temp_file(temporary, output, args.overwrite)
created.append(
{
"path": str(output),
"pages": pages,
"page_count": len(pages),
}
)
finally:
writer.close()
temporary.unlink(missing_ok=True)
return {
"operation": "split",
"input": str(path),
"output_dir": str(destination),
"source_page_count": page_count,
"files": created,
}
def _rotate(args) -> dict[str, Any]:
from pypdf import PdfReader, PdfWriter
path = input_pdf(args.input)
output = output_pdf(args.output, args.overwrite)
if output == path:
raise ValueError("输出文件不能与输入文件相同")
temporary = new_temp_pdf(output)
writer = PdfWriter()
try:
with path.open("rb") as stream:
reader = PdfReader(stream, strict=False)
if reader.is_encrypted:
raise ValueError("PDF 已加密，无法旋转")
page_count = len(reader.pages)
selected = set(parse_page_spec(args.pages, page_count))
for page_number, page in enumerate(reader.pages, start=1):
if page_number in selected:
page.rotate(args.degrees)
writer.add_page(page)
writer.add_metadata(_metadata(reader))
with temporary.open("wb") as output_stream:
writer.write(output_stream)
publish_temp_file(temporary, output, args.overwrite)
finally:
writer.close()
temporary.unlink(missing_ok=True)
return {
"operation": "rotate",
"input": str(path),
"output": str(output),
"page_count": page_count,
"rotated_pages": sorted(selected),
"degrees": args.degrees,
}
def _execute(args) -> dict[str, Any]:
if args.operation == "merge":
return _merge(args)
if args.operation == "split":
return _split(args)
if args.operation == "rotate":
return _rotate(args)
raise ValueError(f"不支持的操作：{args.operation}")
def main(argv: list[str] | None = None) -> int:
arguments = sys.argv[1:] if argv is None else argv
return run_cli(lambda: _execute(_parse_args(arguments)))
if __name__ == "__main__":
raise SystemExit(main())

191
skills/pdf/scripts/render_pdf.py Normal file
View File

@ -0,0 +1,191 @@
#!/usr/bin/env python3
from __future__ import annotations
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Any
from _pdf_common import (
SkillArgumentParser,
input_pdf,
output_directory,
publish_temp_file,
run_cli,
selected_page_window,
)
DEFAULT_DPI = 150
DEFAULT_MAX_PAGES = 10
DEFAULT_TIMEOUT_SECONDS = 180
def _parse_args(argv: list[str]):
parser = SkillArgumentParser(description="使用 Poppler 将 PDF 页面渲染为 PNG")
parser.add_argument("--input", required=True, help="本地 PDF 路径")
parser.add_argument("--output-dir", required=True, help="PNG 输出目录")
parser.add_argument("--start-page", type=int, default=1, help="起始页，1-based")
parser.add_argument("--end-page", type=int, help="结束页，1-based，默认到末页")
parser.add_argument(
"--max-pages",
type=int,
default=DEFAULT_MAX_PAGES,
help=f"单次最多渲染页数，默认 {DEFAULT_MAX_PAGES}",
)
parser.add_argument(
"--dpi",
type=int,
default=DEFAULT_DPI,
help=f"渲染分辨率，默认 {DEFAULT_DPI} DPI",
)
parser.add_argument(
"--timeout",
type=int,
default=DEFAULT_TIMEOUT_SECONDS,
help=f"渲染超时秒数，默认 {DEFAULT_TIMEOUT_SECONDS}",
)
parser.add_argument("--overwrite", action="store_true", help="覆盖已有页面 PNG")
args = parser.parse_args(argv)
if args.dpi < 36 or args.dpi > 600:
raise ValueError("dpi 必须在 36 到 600 之间")
if args.timeout < 1:
raise ValueError("timeout 必须大于 0")
return args
def _page_count(path: Path) -> int:
from pypdf import PdfReader
with path.open("rb") as stream:
reader = PdfReader(stream, strict=False)
if reader.is_encrypted:
raise ValueError("PDF 已加密，无法渲染")
return len(reader.pages)
def _fontconfig_file(pdftoppm: Path) -> Path | None:
for ancestor in pdftoppm.parents:
candidate = (
ancestor
/ "native"
/ "poppler"
/ "poppler"
/ "etc"
/ "fonts"
/ "fonts.conf"
)
if candidate.is_file():
return candidate
for candidate in (
Path("/etc/fonts/fonts.conf"),
Path("/opt/homebrew/etc/fonts/fonts.conf"),
Path("/usr/local/etc/fonts/fonts.conf"),
):
if candidate.is_file():
return candidate
return None
def _render(args) -> dict[str, Any]:
path = input_pdf(args.input)
destination = output_directory(args.output_dir)
page_count = _page_count(path)
start_page, end_page, next_page = selected_page_window(
page_count,
args.start_page,
args.end_page,
args.max_pages,
)
outputs = [
destination / f"page-{page_number:04d}.png"
for page_number in range(start_page, end_page + 1)
]
existing = [path for path in outputs if path.exists()]
if existing and not args.overwrite:
raise FileExistsError(f"页面图片已存在：{existing[0]}")
executable_name = shutil.which("pdftoppm")
if not executable_name:
raise RuntimeError("环境预置的 pdftoppm 不可用")
executable = Path(executable_name).resolve()
with tempfile.TemporaryDirectory(
prefix="pdf-render-",
dir=str(destination),
) as temp_dir_name:
temp_dir = Path(temp_dir_name)
prefix = temp_dir / "page"
font_cache = destination / ".fontconfig-cache"
font_cache.mkdir(parents=True, exist_ok=True)
environment = os.environ.copy()
environment["XDG_CACHE_HOME"] = str(font_cache)
if not environment.get("FONTCONFIG_FILE"):
fontconfig = _fontconfig_file(executable)
if fontconfig is not None:
environment["FONTCONFIG_FILE"] = str(fontconfig)
command = [
str(executable),
"-f",
str(start_page),
"-l",
str(end_page),
"-png",
"-r",
str(args.dpi),
str(path),
str(prefix),
]
try:
completed = subprocess.run(
command,
env=environment,
text=True,
capture_output=True,
timeout=args.timeout,
check=False,
)
except subprocess.TimeoutExpired as exc:
raise RuntimeError(f"PDF 渲染超过 {args.timeout} 秒") from exc
if completed.returncode != 0:
detail = (completed.stderr or completed.stdout or "").strip()[-2000:]
raise RuntimeError(f"PDF 渲染失败：{detail or 'pdftoppm 返回错误'}")
rendered = sorted(
temp_dir.glob("page-*.png"),
key=lambda item: int(item.stem.rsplit("-", 1)[-1]),
)
if len(rendered) != len(outputs):
raise RuntimeError(
f"PDF 渲染页数不正确：预期 {len(outputs)}，实际 {len(rendered)}"
)
for temporary, output in zip(rendered, outputs):
publish_temp_file(temporary, output, args.overwrite)
return {
"input": str(path),
"output_dir": str(destination),
"page_count": page_count,
"start_page": start_page,
"end_page": end_page,
"dpi": args.dpi,
"rendered_count": len(outputs),
"files": [str(path) for path in outputs],
"has_more": next_page is not None,
"next_page": next_page,
}
def main(argv: list[str] | None = None) -> int:
arguments = sys.argv[1:] if argv is None else argv
return run_cli(lambda: _render(_parse_args(arguments)))
if __name__ == "__main__":
raise SystemExit(main())