diff --git a/skills/pdf/SKILL.md b/skills/pdf/SKILL.md index 2dd93a4..86e7a24 100644 --- a/skills/pdf/SKILL.md +++ b/skills/pdf/SKILL.md @@ -1,111 +1,175 @@ --- name: pdf -description: "处理本地 PDF 文件或远程 HTTPS PDF 链接，包括下载、读取、文本与表格提取、页面渲染与视觉审阅、创建、修改、合并、拆分、旋转、表单处理和最终质量校验。当用户提供 .pdf 文件或 HTTPS PDF 地址，或要求生成、编辑、总结、审阅版式重要的 PDF 时使用。" +description: "处理本地 PDF 文件或远程 HTTPS PDF 链接，包括安全下载、元数据与页面检查、分段文本和表格提取、页面 PNG 渲染、从文本创建 PDF、合并、拆分、旋转及最终质量校验。当用户提供 .pdf 文件或 HTTPS PDF 地址，或要求总结、读取、生成、编辑、转换或审阅 PDF 时使用。" --- # PDF 处理 -## 核心原则 +## 强制执行规则 -- 优先进行视觉检查。PDF 的文本提取结果不能代表真实版式，必须把页面渲染为 PNG 后检查。 -- 收到远程 HTTPS PDF 链接时，先下载为本地 PDF，再执行任何读取、编辑或渲染操作。 -- 创建 PDF 时优先使用 `reportlab`；提取文本或表格时使用 `pdfplumber`；读取元数据、合并、拆分、旋转或处理页面结构时使用 `pypdf`。 -- 每次完成影响内容或版式的修改后，重新渲染并检查最新版本。 -- 不覆盖用户提供的源文件。编辑时写入新的输出文件。 +当前智能体不能执行 shell、任意 Python 代码或系统命令。只能通过 `execute_skill_script` 调用本 Skill 中真实存在的固定脚本。 -## 环境约定 +- 只调用下表列出的可执行脚本。 +- 不执行 `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 下载”流程保存到本地。 -2. 校验文件：确认文件存在、大小大于 0、可被 PDF 工具解析，并用 `pdfinfo` 获取页数、页面尺寸、加密状态等信息。 -3. 建立工作目录：把中间文件放在 `tmp/pdfs/<任务名>/`，使用稳定且可读的文件名。 -4. 读取和处理：根据任务使用 `pdfplumber`、`pypdf` 或 `reportlab`，但不要仅依赖文本提取判断页面内容。 -5. 渲染页面：使用 Poppler 把待审阅的 PDF 转成 PNG。 -6. 视觉检查：逐页检查文字、图片、表格、页眉页脚、页码、分页和留白。 -7. 输出结果：把最终 PDF 写入 `output/pdf/`，文件名保持稳定且能表达内容。 -8. 清理中间文件：交付完成后仅删除本次任务对应的 `tmp/pdfs/<任务名>/`，保留最终产物。 +1. 为任务选择简短目录名，把中间文件放在 `tmp/pdfs/<任务名>/`。 +2. 远程 HTTPS 链接先调用 `download_pdf.py`；本地文件直接进入下一步。 +3. 调用 `inspect_pdf.py` 检查文件。遇到加密 PDF 且任务需要正文时，向用户索取密码，不要绕过加密。 +4. 阅读或总结时，循环调用 `extract_text.py` 直至 `has_more` 为 `false`；需要表格时再调用 `extract_tables.py`。 +5. 调用 `render_pdf.py` 检查首页、复杂页面、无文本页面和用户关心的页面。创建或修改 PDF 时分批渲染全部页面。 +6. 使用提取文本和渲染结果完成回答；明确区分原文内容与基于版式或图表的推断。 +7. 创建或修改后的最终 PDF 写入 `output/pdf/`，重新执行检查、文本提取和全部页面渲染。 +8. 最终产物位于临时目录之外且不再需要缓存时，调用 `cleanup_pdf_temp.py` 清理本次任务目录。 -## 远程 PDF 下载 +## 下载远程 PDF -只接受 HTTPS 地址。必须使用本 Skill 自带的 `scripts/download_pdf.py` 下载，不要使用 `curl`、shell 脚本或临时编写的下载逻辑。脚本会自动创建目标目录、流式下载、限制重定向协议、控制文件大小，并使用 `pypdf` 校验下载内容。 +只接受 HTTPS 地址。完整保留 URL 及查询参数，不在回复、日志摘要或文件名中复述敏感参数。 + +调用 `scripts/download_pdf.py`： ```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`。 -- `--max-bytes <字节数>`：最大下载大小，默认 `104857600`（100 MiB）。 -- `--overwrite`：覆盖已存在的目标文件。 +- `--timeout <秒>`：默认 `60`。 +- `--max-bytes <字节数>`：默认 `104857600`（100 MiB）。 +- `--overwrite`：仅在目标是本次任务生成的缓存时使用。 -脚本成功时退出码为 `0`，并输出： +脚本会创建父目录、流式下载、阻止 HTTPS 重定向降级到 HTTP，并验证 PDF。成功结果包含 `path`、`size_bytes`、`page_count` 和 `encrypted`。 -```json -{ - "ok": true, - "path": "/absolute/path/to/source.pdf", - "size_bytes": 123456, - "page_count": 10, - "encrypted": false -} +## 检查 PDF + +调用 `scripts/inspect_pdf.py`： + +```text +--input 'tmp/pdfs/<任务名>/source.pdf' ``` -脚本失败时退出码为非 `0`，并输出 `{"ok": false, "error": "具体错误"}`。只有收到 `ok: true` 后才能继续读取或渲染该文件。 +使用返回的 `page_count`、`encrypted`、`metadata`、`page_layouts` 和 `form_field_count` 判断后续处理方式。不要直接调用 `pdfinfo`。 -遵守以下规则： +## 提取正文 -- 跟随 HTTPS 重定向，但不允许降级到 HTTP。 -- 不根据 URL 后缀或响应的 `Content-Type` 单独判断文件类型，以实际 PDF 解析结果为准。 -- 如果下载结果是 HTML、登录页、错误页或无法解析的内容，停止后续处理并明确说明链接需要授权或真实 PDF 下载地址。 -- 如果链接包含签名、令牌或其他敏感查询参数，不在最终回复、日志摘要、错误信息或生成的文件名中复述这些参数；下载脚本也不会在结果中回显 URL。 -- 如果 PDF 已加密且任务需要密码，先向用户索取密码；不得尝试绕过加密。 +首次调用 `scripts/extract_text.py`： -## 读取与审阅 - -- 使用 `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' +```text +--input 'tmp/pdfs/<任务名>/source.pdf' ``` -对于文字较小、图表密集或 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 ` +- `--font-path ` +- `--font-size <字号>` +- `--margin ` +- `--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。 diff --git a/skills/pdf/scripts/_pdf_common.py b/skills/pdf/scripts/_pdf_common.py new file mode 100644 index 0000000..1c29cca --- /dev/null +++ b/skills/pdf/scripts/_pdf_common.py @@ -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 diff --git a/skills/pdf/scripts/cleanup_pdf_temp.py b/skills/pdf/scripts/cleanup_pdf_temp.py new file mode 100644 index 0000000..5d865d3 --- /dev/null +++ b/skills/pdf/scripts/cleanup_pdf_temp.py @@ -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()) diff --git a/skills/pdf/scripts/create_pdf.py b/skills/pdf/scripts/create_pdf.py new file mode 100644 index 0000000..33cb378 --- /dev/null +++ b/skills/pdf/scripts/create_pdf.py @@ -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 = "
".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()) diff --git a/skills/pdf/scripts/extract_tables.py b/skills/pdf/scripts/extract_tables.py new file mode 100644 index 0000000..7eb798b --- /dev/null +++ b/skills/pdf/scripts/extract_tables.py @@ -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()) diff --git a/skills/pdf/scripts/extract_text.py b/skills/pdf/scripts/extract_text.py new file mode 100644 index 0000000..ed8030e --- /dev/null +++ b/skills/pdf/scripts/extract_text.py @@ -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()) diff --git a/skills/pdf/scripts/inspect_pdf.py b/skills/pdf/scripts/inspect_pdf.py new file mode 100644 index 0000000..6a91064 --- /dev/null +++ b/skills/pdf/scripts/inspect_pdf.py @@ -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()) diff --git a/skills/pdf/scripts/manage_pdf.py b/skills/pdf/scripts/manage_pdf.py new file mode 100644 index 0000000..7474fab --- /dev/null +++ b/skills/pdf/scripts/manage_pdf.py @@ -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()) diff --git a/skills/pdf/scripts/render_pdf.py b/skills/pdf/scripts/render_pdf.py new file mode 100644 index 0000000..886a451 --- /dev/null +++ b/skills/pdf/scripts/render_pdf.py @@ -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())