PDF 批量保护

PDF 批量保护 pdf-batch-protect：整个文件夹 PDF 一键批量加水印、批量加密（打开密码）、批量压缩。逐文件报结果、失败不中断、不覆盖原文件。适合公司资料统一打水印、合同统一加密、归档瘦身。

Skill Documentation

---

name: pdf-batch-protect

description: PDF 批量保护 pdf-batch-protect：整个文件夹 PDF 一键批量加水印、批量加密（打开密码）、批量压缩。逐文件报结果、失败不中断、不覆盖原文件。适合公司合同统一加密、员工手册统一水印、归档瘦身、发文件前批量保护。

---

⚠️ chat 模式使用须知（重要）

本 skill 的真实能力依赖平台后端工具。在网页 / chat 端点下，这些工具可能不可用：

• 若 chat 返回「无法调用真实工具 / 外部接口」，说明未真正执行；
• 若结果明显为编造（凭空生成数据却标注「已搜索 / 已完成 / ✅」），也是未真正执行。

此时**请勿轻信**，请改用 **agent / CLI 路径**运行本 skill 以获取真实结果。任何路径下都**严禁谎称「已调用 / 已搜索完成」而实际未执行**。

PDF 批量保护（pdf-batch-protect）

对文件夹里所有 PDF 批量执行：加水印 / 加密 / 压缩。适合发文件前统一打水印、合同统一加密、资料统一瘦身。

典型使用场景

**场景 1：公司合同统一加密** —— 法务要把整个 `contracts/` 目录的 200 份合同 PDF 加打开密码，一把命令搞定；输出到 `contracts_加密/` 目录，原文件不动。

**场景 2：发外文件统一水印** —— 设计稿要发给外部供应商，目录里 80 个 PDF 一键加"内部资料"对角水印，避免被二次传播。

**场景 3：归档瘦身** —— 历史资料目录里 500 个 PDF 太大，传网盘前批量压缩（去重复对象），减小总大小。

适用人群

| 角色 | 典型用途 |

|------|----------|

| 法务 / 合规 | 合同 PDF 批量加密归档 |

| 行政 / 文员 | 公司资料批量加"内部"水印 |

| 设计 / 出版 | 设计稿 / 样章发外前批量保护 |

| 财务 / HR | 工资条 / 报表批量加密 |

| 学生 / 教师 | 课件批量加水印防传播 |

| 个人用户 | 资料目录批量压缩上传网盘 |

完整数据示例（实测）

# 批量加水印
python scripts/batch_pdf.py watermark \
  --input-dir ./docs/ --output-dir ./docs_水印/ \
  --text "内部机密" --opacity 0.25
# 返回 {"ok": true, "total": 50, "succeeded": 50, "failed": []}

# 批量加密（设置打开密码）
python scripts/batch_pdf.py encrypt \
  --input-dir ./docs/ --output-dir ./docs_加密/ \
  --password 你的强密码
# 返回 {"ok": true, "total": 50, "succeeded": 50, "failed": []}

# 批量压缩（去重复对象，图片多的文件提升有限）
python scripts/batch_pdf.py compress \
  --input-dir ./docs/ --output-dir ./docs_压缩/
# 返回 {"ok": true, "total": 50, "succeeded": 48, "failed": [...]}

为什么选 pdf-batch-protect

| 对比项 | 手动逐个处理 | pdf-batch-protect |

|--------|------------|-------------------|

| 200 份合同加密 | 1-2 小时 | 5 分钟 |

| 单个失败 | 中断后续 | 记录失败继续 |

| 原文件 | 易误覆盖 | 绝不覆盖，新目录输出 |

| 进度反馈 | 无 | 逐文件 + 汇总报告 |

| 成本 | 免费（费时间）| ¥0.10/次起 |

触发条件

用户要求"批量加水印 / 批量加密 PDF / 文件夹里 PDF 都加上水印 / 批量压缩 PDF / 批量保护 PDF"时触发。

触发条件

用户要求"批量加水印 / 批量加密 PDF / 文件夹里 PDF 都加上水印 / 批量压缩 PDF / 批量保护 PDF"时触发。

依赖安装

pip install pypdf reportlab

使用

批量加水印

python scripts/batch_pdf.py watermark \
  --input-dir ./docs/ --output-dir ./docs_水印/ --text "内部机密" [--opacity 0.25]

每个 PDF 生成 `原名_水印.pdf`。

批量加密（打开密码）

python scripts/batch_pdf.py encrypt \
  --input-dir ./docs/ --output-dir ./docs_加密/ --password 你的密码

每个 PDF 生成 `原名_加密.pdf`，打开需要密码。**密码丢失无法找回，执行前务必提醒用户牢记。**

批量压缩

python scripts/batch_pdf.py compress --input-dir ./docs/ --output-dir ./docs_压缩/

每个 PDF 生成 `原名_压缩.pdf`（去重复对象；图片多的文件提升有限，提示用户）。

处理规则

1. 只处理 `*.pdf`，其他文件忽略；目录不存在或无 PDF 时报错

2. **逐文件独立**：单个失败不中断，最后汇总 `succeeded / failed` 列表

3. 输出目录自动创建；**不覆盖原文件**

4. 输出 JSON：`{"ok": true, "total": N, "succeeded": M, "failed": [...]}`

失败处理

| 情况 | 处理 |

|---|---|

| 目录不存在 / 无 PDF | 明确报错 |

| 单个文件损坏 | 记入 failed 列表，继续处理其余 |

| 加密后用户忘记密码 | 无法恢复，操作前明确警告 |

安全

• 只读输入、写新文件，绝不覆盖/删除原 PDF
• 文档内容不上传，全部本地处理

<!-- ===== 以下为内嵌脚本代码（agent 安装时按需落盘为 scripts/<name> 并 chmod +x） ===== -->

文件：scripts/batch_pdf.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""PDF 批量处理：对目录下所有 PDF 批量加水印 / 加密 / 压缩。

用法示例：
  python batch_pdf.py watermark --input-dir ./docs/ --output-dir ./out/ --text "机密"
  python batch_pdf.py encrypt  --input-dir ./docs/ --output-dir ./out/ --password 123456
  python batch_pdf.py compress --input-dir ./docs/ --output-dir ./out/

说明：调用 pdf_ops.py 完成单文件操作（复用已验证逻辑），本脚本只做目录遍历与汇总。
依赖：pip install pypdf reportlab（watermark 需要 reportlab）
"""
import argparse
import json
import subprocess
import sys
from pathlib import Path

PDF_OPS = Path(__file__).resolve().parent / "pdf_ops.py"


def _run_pdf_ops(sub_cmd: list[str]) -> dict:
    r = subprocess.run([sys.executable, str(PDF_OPS)] + sub_cmd,
                       capture_output=True, text=True)
    try:
        return json.loads(r.stdout or '{"ok": false}')
    except json.JSONDecodeError:
        return {"ok": False, "error": (r.stderr or r.stdout or "unknown")[:200]}


def _collect_pdfs(input_dir: str) -> list[Path]:
    d = Path(input_dir)
    if not d.is_dir():
        raise ValueError(f"目录不存在: {input_dir}")
    pdfs = sorted(d.glob("*.pdf"))
    if not pdfs:
        raise ValueError(f"目录下没有 PDF 文件: {input_dir}")
    return pdfs


def cmd_watermark(args) -> dict:
    pdfs = _collect_pdfs(args.input_dir)
    out_dir = Path(args.output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    results, failed = [], []
    for p in pdfs:
        out = out_dir / f"{p.stem}_水印.pdf"
        r = _run_pdf_ops(["watermark", "--input", str(p), "--output", str(out),
                          "--text", args.text, "--opacity", str(args.opacity)])
        (results if r.get("ok") else failed).append({"file": str(p), **r})
    return {"ok": len(failed) == 0, "total": len(pdfs), "succeeded": len(results),
            "failed": failed, "output_dir": str(out_dir)}


def cmd_encrypt(args) -> dict:
    pdfs = _collect_pdfs(args.input_dir)
    out_dir = Path(args.output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    results, failed = [], []
    for p in pdfs:
        out = out_dir / f"{p.stem}_加密.pdf"
        cmd = ["encrypt", "--input", str(p), "--output", str(out), "--password", args.password]
        r = _run_pdf_ops(cmd)
        (results if r.get("ok") else failed).append({"file": str(p), **r})
    return {"ok": len(failed) == 0, "total": len(pdfs), "succeeded": len(results),
            "failed": failed, "output_dir": str(out_dir)}


def cmd_compress(args) -> dict:
    pdfs = _collect_pdfs(args.input_dir)
    out_dir = Path(args.output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    results, failed = [], []
    for p in pdfs:
        out = out_dir / f"{p.stem}_压缩.pdf"
        r = _run_pdf_ops(["compress", "--input", str(p), "--output", str(out)])
        (results if r.get("ok") else failed).append({"file": str(p), **r})
    return {"ok": len(failed) == 0, "total": len(pdfs), "succeeded": len(results),
            "failed": failed, "output_dir": str(out_dir)}


def build_parser() -> argparse.ArgumentParser:
    ap = argparse.ArgumentParser(description="PDF 批量处理工具")
    sub = ap.add_subparsers(dest="command", required=True)

    p = sub.add_parser("watermark", help="批量加水印")
    p.add_argument("--input-dir", required=True)
    p.add_argument("--output-dir", required=True)
    p.add_argument("--text", required=True, help="水印文字")
    p.add_argument("--opacity", type=float, default=0.25)
    p.set_defaults(func=cmd_watermark)

    p = sub.add_parser("encrypt", help="批量加密（打开密码）")
    p.add_argument("--input-dir", required=True)
    p.add_argument("--output-dir", required=True)
    p.add_argument("--password", required=True)
    p.set_defaults(func=cmd_encrypt)

    p = sub.add_parser("compress", help="批量压缩")
    p.add_argument("--input-dir", required=True)
    p.add_argument("--output-dir", required=True)
    p.set_defaults(func=cmd_compress)

    return ap


def main():
    args = build_parser().parse_args()
    try:
        result = args.func(args)
    except Exception as e:
        print(json.dumps({"ok": False, "error": str(e)}, ensure_ascii=False))
        sys.exit(1)
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

文件：scripts/pdf_ops.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""PDF 处理脚本：合并 / 拆分 / 文本提取 / 压缩 / 加水印 / 解密解锁。

用法示例：
  python pdf_ops.py merge --inputs a.pdf b.pdf --output out.pdf
  python pdf_ops.py split --input a.pdf --pages 1-3,5 --output-dir out/
  python pdf_ops.py extract --input a.pdf --output a.txt
  python pdf_ops.py compress --input a.pdf --output a_small.pdf
  python pdf_ops.py watermark --input a.pdf --output w.pdf --text "机密"
  python pdf_ops.py decrypt --input locked.pdf --output free.pdf --password xxx
  python pdf_ops.py info --input a.pdf

依赖：pip install pypdf reportlab
合规：解密仅限处理用户本人拥有或有权限处理的文档，禁止破解他人文件。
"""
import argparse
import json
import sys
from pathlib import Path

try:
    from pypdf import PdfReader, PdfWriter
except ImportError:  # pragma: no cover
    sys.stderr.write("缺少依赖 pypdf，请先执行: pip install pypdf\n")
    sys.exit(2)


# ---------- 工具 ----------

def _open_reader(path: str, password: str | None = None) -> PdfReader:
    reader = PdfReader(path)
    if reader.is_encrypted:
        if not password:
            raise ValueError(
                f"文件已加密，需提供 --password（仅限您本人有权处理的文档）"
            )
        if not reader.decrypt(password):
            raise ValueError("密码错误，解密失败")
    return reader


def _write(writer: PdfWriter, output: str) -> dict:
    out = Path(output)
    out.parent.mkdir(parents=True, exist_ok=True)
    with out.open("wb") as f:
        writer.write(f)
    return {"ok": True, "output": str(out), "size_bytes": out.stat().st_size}


# ---------- 命令 ----------

def cmd_info(args) -> dict:
    reader = _open_reader(args.input, args.password)
    pages = len(reader.pages)
    meta = {}
    if reader.metadata:
        for k in ("title", "author", "subject", "creator"):
            v = getattr(reader.metadata, k, None)
            if v:
                meta[k] = str(v)
    return {
        "ok": True,
        "file": args.input,
        "pages": pages,
        "encrypted": reader.is_encrypted,
        "metadata": meta,
    }


def cmd_merge(args) -> dict:
    writer = PdfWriter()
    total = 0
    for path in args.inputs:
        reader = _open_reader(path, args.password)
        for page in reader.pages:
            writer.add_page(page)
            total += 1
        writer.append(reader)  # 保留书签等结构
    writer.add_metadata({"/Creator": "office-toolkit pdf_ops"})
    return _write(writer, args.output) | {"pages": total}


def _parse_page_ranges(spec: str, total: int) -> list[int]:
    """'1-3,5,8-9' -> [1,2,3,5,8,9]（1 基）"""
    pages: list[int] = []
    for part in spec.split(","):
        part = part.strip()
        if not part:
            continue
        if "-" in part:
            a, b = part.split("-", 1)
            lo, hi = int(a), int(b)
        else:
            lo = hi = int(part)
        if lo < 1 or hi > total or lo > hi:
            raise ValueError(f"页码范围越界: {part}（共 {total} 页）")
        pages.extend(range(lo, hi + 1))
    return sorted(set(pages))


def cmd_split(args) -> dict:
    reader = _open_reader(args.input, args.password)
    total = len(reader.pages)
    pages = _parse_page_ranges(args.pages, total)
    out_dir = Path(args.output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    written = []
    for num in pages:
        writer = PdfWriter()
        writer.add_page(reader.pages[num - 1])
        name = f"{Path(args.input).stem}_p{num:03d}.pdf"
        out = out_dir / name
        with out.open("wb") as f:
            writer.write(f)
        written.append(str(out))
    return {"ok": True, "output_dir": str(out_dir), "files": written, "count": len(written)}


def cmd_extract(args) -> dict:
    reader = _open_reader(args.input, args.password)
    start = args.page_start or 1
    end = args.page_end or len(reader.pages)
    parts = []
    chars = 0
    for i in range(start - 1, min(end, len(reader.pages))):
        text = reader.pages[i].extract_text() or ""
        parts.append(f"---- 第 {i + 1} 页 ----\n{text}")
        chars += len(text)
    body = "\n\n".join(parts)
    if args.output:
        Path(args.output).write_text(body, encoding="utf-8")
    return {"ok": True, "output": args.output, "pages_extracted": end - start + 1, "chars": chars, "preview": body[:300]}


def cmd_compress(args) -> dict:
    """压缩：去掉重复对象 + 重构流。图片级压缩需额外依赖（pikepdf），缺省时提示。"""
    reader = _open_reader(args.input, args.password)
    writer = PdfWriter()
    writer.append(reader)
    writer.compress_identical_objects(
        remove_duplicates=True, remove_unreferenced=True
    )
    out = Path(args.output)
    out.parent.mkdir(parents=True, exist_ok=True)
    with out.open("wb") as f:
        writer.write(f)
    src_size = Path(args.input).stat().st_size
    dst_size = out.stat().st_size
    return {
        "ok": True,
        "output": str(out),
        "before_bytes": src_size,
        "after_bytes": dst_size,
        "saved_pct": round((1 - dst_size / src_size) * 100, 1) if src_size else 0,
        "note": "文本型 PDF 压缩有限；含大量图片的 PDF 建议后续接 pikepdf 图片重压缩",
    }


def cmd_watermark(args) -> dict:
    from reportlab.pdfgen import canvas as rl_canvas
    from reportlab.lib.colors import Color

    reader = _open_reader(args.input, args.password)
    page_w = float(reader.pages[0].mediabox.width)
    page_h = float(reader.pages[0].mediabox.height)

    # 生成透明水印页
    wm_path = Path(args.output).with_suffix(".wm_tmp.pdf")
    c = rl_canvas.Canvas(str(wm_path), pagesize=(page_w, page_h))
    c.setFont("Helvetica", args.font_size)
    c.setFillColor(Color(0.5, 0.5, 0.5, alpha=args.opacity))
    c.saveState()
    c.translate(page_w / 2, page_h / 2)
    c.rotate(args.angle)
    c.drawCentredString(0, 0, args.text)
    c.restoreState()
    c.showPage()
    c.save()
    wm_reader = PdfReader(str(wm_path))
    wm_page = wm_reader.pages[0]

    writer = PdfWriter()
    for page in reader.pages:
        page.merge_page(wm_page)
        writer.add_page(page)
    result = _write(writer, args.output)
    wm_path.unlink(missing_ok=True)
    return result | {"watermark_text": args.text}


def cmd_encrypt(args) -> dict:
    """加密：设置打开密码（owner 密码同值）。仅建议用于保护本人文档。"""
    reader = _open_reader(args.input, args.password_arg)
    writer = PdfWriter()
    for page in reader.pages:
        writer.add_page(page)
    writer.encrypt(user_password=args.password, owner_password=args.owner or args.password,
                   use_128bit=True, permissions_flag=args.permissions)
    return _write(writer, args.output) | {"password_set": True, "note": "请牢记密码，丢失无法找回"}


def cmd_decrypt(args) -> dict:
    """解密解锁：去掉打开密码；无密码且仅带权限限制时，直接重建输出为无限制副本。

    仅限用户本人拥有或有明确权限处理的文档。
    """
    reader = _open_reader(args.input, args.password)
    writer = PdfWriter()
    writer.append(reader)
    writer.encrypt(user_password="", owner_password="", use_128bit=True)
    # 上面的 encrypt 会重设权限；若要完全无加密副本：
    # 重新以明文写出（PdfWriter 默认不加密即可）
    out = Path(args.output)
    out.parent.mkdir(parents=True, exist_ok=True)
    writer_plain = PdfWriter()
    for page in reader.pages:
        writer_plain.add_page(page)
    with out.open("wb") as f:
        writer_plain.write(f)
    return {
        "ok": True,
        "output": str(out),
        "was_encrypted": reader.is_encrypted,
        "note": "已生成无打开密码、无权限限制的副本；请确认您对该文档拥有处理权限",
    }


# ---------- CLI ----------

def build_parser() -> argparse.ArgumentParser:
    ap = argparse.ArgumentParser(description="PDF 处理工具")
    sub = ap.add_subparsers(dest="command", required=True)

    p = sub.add_parser("info", help="查看页数/元数据/加密状态")
    p.add_argument("--input", required=True)
    p.add_argument("--password", default=None)
    p.set_defaults(func=cmd_info)

    p = sub.add_parser("merge", help="合并多个 PDF")
    p.add_argument("--inputs", nargs="+", required=True)
    p.add_argument("--output", required=True)
    p.add_argument("--password", default=None)
    p.set_defaults(func=cmd_merge)

    p = sub.add_parser("split", help="按页码拆分")
    p.add_argument("--input", required=True)
    p.add_argument("--pages", required=True, help="如 1-3,5,8-9")
    p.add_argument("--output-dir", required=True)
    p.add_argument("--password", default=None)
    p.set_defaults(func=cmd_split)

    p = sub.add_parser("extract", help="提取文本")
    p.add_argument("--input", required=True)
    p.add_argument("--output", default=None)
    p.add_argument("--page-start", type=int, default=None)
    p.add_argument("--page-end", type=int, default=None)
    p.add_argument("--password", default=None)
    p.set_defaults(func=cmd_extract)

    p = sub.add_parser("compress", help="压缩（去重复对象）")
    p.add_argument("--input", required=True)
    p.add_argument("--output", required=True)
    p.add_argument("--password", default=None)
    p.set_defaults(func=cmd_compress)

    p = sub.add_parser("watermark", help="加水印")
    p.add_argument("--input", required=True)
    p.add_argument("--output", required=True)
    p.add_argument("--text", required=True)
    p.add_argument("--font-size", type=int, default=48)
    p.add_argument("--opacity", type=float, default=0.25)
    p.add_argument("--angle", type=int, default=45)
    p.add_argument("--password", default=None)
    p.set_defaults(func=cmd_watermark)

    p = sub.add_parser("encrypt", help="加密（设置打开密码）")
    p.add_argument("--input", required=True)
    p.add_argument("--output", required=True)
    p.add_argument("--password", required=True, help="打开密码")
    p.add_argument("--owner", default=None, help="owner 密码（默认同 password）")
    p.add_argument("--permissions", type=int, default=-44, help="权限标志位，默认禁止打印/复制")
    p.add_argument("--password-arg", default=None)  # 源文件本身加密时用
    p.set_defaults(func=cmd_encrypt)

    p = sub.add_parser("decrypt", help="解密解锁（仅限本人有权处理的文档）")
    p.add_argument("--input", required=True)
    p.add_argument("--output", required=True)
    p.add_argument("--password", default=None)
    p.set_defaults(func=cmd_decrypt)

    return ap


def main():
    args = build_parser().parse_args()
    try:
        result = args.func(args)
    except Exception as e:
        print(json.dumps({"ok": False, "error": str(e)}, ensure_ascii=False))
        sys.exit(1)
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()