PDF 图片往返转换：Mask 坐标保护

用途

保存 2026-11-03 最终版 Mask 坐标保护 的 Python 脚本（来自前次 PDF 往返转换需求），方便后续复用与改进。此 Issue 直接包含完整源码，不只是需求描述。

核心需求

• 弹窗连续添加多个输入目录，按队列顺序处理；目录内递归扫描图片。
• 每张图依次执行 图像 → 单页 PDF → PNG；允许 .jpg/.jpeg/.png/.bmp/.tif/.tiff/.webp。
• 输出到输入目录同级的 <目录名>_PDF、<目录名>_PDF2IMG，保留原有相对子目录结构。
• 输出图片文件名在原名后追加 _p2i，如 xxx.png → xxx_p2i.png。
• 默认 DPI=300，PDF 页面尺寸计算：W_pt = W_px * 72 / DPI，H_pt = H_px * 72 / DPI。
• 不缩放、不插值、不旋转、不裁切、不留白、不平移。PDF 渲染尺寸与原图不同则报错，不输出错误坐标图片。
• 每个输入目录生成 尺寸校验报告.csv，记录源图宽高、PDF 页面大小、渲染结果与失败原因。
• 多个输入目录重叠时，同一原始文件只处理一次。

安装与运行

pip install pymupdf pillow
python image_pdf_roundtrip_multi_dirs_p2i_mask_safe.py

重要边界： 这个脚本保障的是最终 PNG 的像素宽高/坐标网格与原图一致，而不是 RGB 数值、灰度或分割 Mask 标签值逐点相同。如果 Mask 像素值必须完全无损，不应让 Mask 经历 PDF 渲染往返。

另外，DPI 影响 PDF 页面物理尺寸；本脚本的严格模式不会为匹配尺寸偷偷 resize，遇到渲染取整差异会判失败。

完整脚本：image_pdf_roundtrip_multi_dirs_p2i_mask_safe.py

# -*- coding: utf-8 -*-
"""
严格 Mask 坐标保护版：多目录排队处理
图像 -> PDF -> 图像（输出文件名追加 _p2i）

安装依赖：
    pip install pymupdf pillow

运行：
    python image_pdf_roundtrip_multi_dirs_p2i_mask_safe.py

保护规则：
1. 不调用 resize，不做 LANCZOS/BICUBIC 等插值缩放。
2. 不做 EXIF 自动旋转，不旋转 PDF 页面。
3. 不裁切、不留白、不平移，图像直接铺满与原图比例一致的 PDF 页面。
4. PDF 渲染后的宽高只要与原图不完全一致，就判定失败：
   不生成错误的 p2i 图片，并将原因写入 CSV。
5. 保留多目录队列、递归扫描、重复源文件跳过、_p2i 命名。
6. 每个目录分别生成 PDF、PDF2IMG 输出目录和尺寸校验报告。

重要：
- 该脚本保护的是像素坐标网格与宽高，不保证 PDF 栅格化前后的每个像素颜色值
  或 Mask 灰度值逐点完全相同。若需要像素值绝对无损，Mask 不应经过 PDF 往返。
- 默认 DPI=300。修改 DPI 会改变 PDF 页面物理尺寸，但本脚本会要求渲染像素宽高
  与源图一致；若当前 PDF 渲染器产生尺寸取整差异，则记录失败，不擅自缩放修正。
"""

from __future__ import annotations

import csv
import io
import sys
import traceback
from pathlib import Path
import tkinter as tk
from tkinter import filedialog, messagebox

import fitz  # PyMuPDF
from PIL import Image

# ===================== 可修改配置 =====================
DPI = 300
SUPPORTED_EXTS = {
    ".jpg", ".jpeg", ".png", ".bmp", ".tif", ".tiff", ".webp"
}
# =====================================================


def image_to_png_bytes(image_path: Path, dpi: int):
    """
    读取源图并编码为 PNG，保持像素宽高。
    不执行 EXIF transpose，不改变图像方向。
    对常见 8-bit 模式尽量保留原模式，避免不必要的色彩转换。
    """
    with Image.open(image_path) as im:
        im.load()
        width, height = im.size
        source_mode = im.mode

        # 高位深度/浮点模式不能静默降位，否则可能破坏 Mask 标签值。
        if source_mode in ("I", "F", "I;16", "I;16B", "I;16L"):
            raise ValueError(
                f"源图模式为 {source_mode}，可能是高位深度/浮点图像；"
                "为避免 Mask 数值被降位，本脚本拒绝自动转换。"
            )

        has_alpha = (
            "A" in im.getbands()
            or "transparency" in im.info
        )
        icc_profile = im.info.get("icc_profile")
        transparency = im.info.get("transparency")

        # 直接保留 PNG 常见模式；LA 转 RGBA 是为了 PDF 透明通道兼容。
        if source_mode in ("1", "L", "P", "RGB", "RGBA"):
            converted = im.copy()
        elif source_mode == "LA":
            converted = im.convert("RGBA")
        else:
            converted = im.convert("RGBA" if has_alpha else "RGB")

        buf = io.BytesIO()
        save_kwargs = {
            "format": "PNG",
            "dpi": (dpi, dpi),
        }
        if icc_profile:
            save_kwargs["icc_profile"] = icc_profile
        if transparency is not None and converted.mode == "P":
            save_kwargs["transparency"] = transparency

        converted.save(buf, **save_kwargs)
        return buf.getvalue(), width, height, has_alpha, source_mode


def convert_one(
    image_path: Path,
    input_root: Path,
    pdf_root: Path,
    image_root: Path,
    dpi: int,
):
    relative = image_path.relative_to(input_root)

    pdf_path = (pdf_root / relative).with_suffix(".pdf")
    output_path = (image_root / relative).with_name(
        relative.stem + "_p2i.png"
    )

    pdf_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.parent.mkdir(parents=True, exist_ok=True)

    png_bytes, original_w, original_h, has_alpha, source_mode = (
        image_to_png_bytes(image_path, dpi)
    )

    # 根据源图像素宽高和 DPI 反算页面物理尺寸（PDF point）。
    # 1 inch = 72 point，因此页面尺寸 = 像素数 * 72 / DPI。
    page_w_pt = original_w * 72.0 / dpi
    page_h_pt = original_h * 72.0 / dpi

    # 第一步：图像 -> PDF。页面与图像宽高比完全一致，图片铺满页面。
    pdf_doc = fitz.open()
    page = pdf_doc.new_page(width=page_w_pt, height=page_h_pt)
    page.set_rotation(0)
    page.insert_image(
        page.rect,
        stream=png_bytes,
        keep_proportion=False,
        overlay=True,
    )
    pdf_doc.save(str(pdf_path), garbage=4, deflate=True)
    pdf_doc.close()

    # 第二步：PDF -> 图像。相同 DPI 渲染，不进行任何尺寸修正。
    with fitz.open(str(pdf_path)) as read_doc:
        if read_doc.page_count != 1:
            raise RuntimeError(
                f"生成的 PDF 页数异常：{read_doc.page_count}，预期 1 页"
            )

        read_page = read_doc[0]
        if read_page.rotation != 0:
            raise RuntimeError(
                f"PDF 页面存在旋转：{read_page.rotation}°，拒绝继续"
            )

        pix = read_page.get_pixmap(
            dpi=dpi,
            alpha=has_alpha,
            annots=False,
        )

        rendered_w = pix.width
        rendered_h = pix.height

        # 严格策略：尺寸只要不一致就失败，不缩放、不裁切、不补边。
        if (rendered_w, rendered_h) != (original_w, original_h):
            raise RuntimeError(
                "PDF 渲染尺寸与源图不一致；为保护 Mask 坐标，"
                "已拒绝缩放/裁切/补边。"
                f"源图={original_w}x{original_h}，"
                f"渲染={rendered_w}x{rendered_h}，DPI={dpi}。"
            )

        rendered_bytes = pix.tobytes("png")

    # 只重新编码 PNG 并写入 DPI 元数据，不改变像素矩阵。
    with Image.open(io.BytesIO(rendered_bytes)) as rendered:
        rendered.load()
        if rendered.size != (original_w, original_h):
            raise RuntimeError(
                f"PNG 解码后尺寸异常：{rendered.width}x{rendered.height}"
            )
        rendered.save(
            str(output_path),
            format="PNG",
            dpi=(dpi, dpi),
        )

    with Image.open(output_path) as check:
        final_w, final_h = check.size

    if (final_w, final_h) != (original_w, original_h):
        # 不保留尺寸错误的结果文件。
        try:
            output_path.unlink(missing_ok=True)
        except Exception:
            pass
        raise RuntimeError(
            f"最终文件尺寸校验失败：源图={original_w}x{original_h}，"
            f"输出={final_w}x{final_h}"
        )

    return {
        "源文件": str(image_path),
        "PDF文件": str(pdf_path),
        "输出图像": str(output_path),
        "源图模式": source_mode,
        "DPI": dpi,
        "原图宽": original_w,
        "原图高": original_h,
        "PDF页面宽(pt)": round(page_w_pt, 6),
        "PDF页面高(pt)": round(page_h_pt, 6),
        "渲染宽": rendered_w,
        "渲染高": rendered_h,
        "最终宽": final_w,
        "最终高": final_h,
        "尺寸一致": "是",
        "状态": "成功",
        "错误信息": "",
    }


def process_directory(input_root: Path, globally_processed: set[str]):
    pdf_root = input_root.parent / f"{input_root.name}_PDF"
    image_root = input_root.parent / f"{input_root.name}_PDF2IMG"

    image_files = sorted(
        p for p in input_root.rglob("*")
        if p.is_file() and p.suffix.lower() in SUPPORTED_EXTS
    )

    report_rows = []
    success_count = 0
    failed_count = 0
    skipped_count = 0

    print("\n" + "=" * 90)
    print(f"开始目录：{input_root}")
    print(f"图片数量：{len(image_files)}")
    print(f"PDF 输出：{pdf_root}")
    print(f"图像输出：{image_root}")
    print("=" * 90)

    pdf_root.mkdir(parents=True, exist_ok=True)
    image_root.mkdir(parents=True, exist_ok=True)

    for index, image_path in enumerate(image_files, 1):
        source_key = str(image_path.resolve()).casefold()

        if source_key in globally_processed:
            skipped_count += 1
            print(f"[跳过重复 {index}/{len(image_files)}] {image_path}")
            continue

        globally_processed.add(source_key)
        print(f"[{index}/{len(image_files)}] {image_path}")

        try:
            row = convert_one(
                image_path,
                input_root,
                pdf_root,
                image_root,
                DPI,
            )
            report_rows.append(row)
            success_count += 1
            print(
                f"  成功：{row['原图宽']}x{row['原图高']} "
                f"-> {row['最终宽']}x{row['最终高']} "
                f"| {row['输出图像']}"
            )
        except Exception as exc:
            failed_count += 1
            report_rows.append({
                "源文件": str(image_path),
                "PDF文件": "",
                "输出图像": "",
                "源图模式": "",
                "DPI": DPI,
                "原图宽": "",
                "原图高": "",
                "PDF页面宽(pt)": "",
                "PDF页面高(pt)": "",
                "渲染宽": "",
                "渲染高": "",
                "最终宽": "",
                "最终高": "",
                "尺寸一致": "否",
                "状态": "失败",
                "错误信息": f"{type(exc).__name__}: {exc}",
            })
            print(f"  失败：{type(exc).__name__}: {exc}")

    report_path = image_root / "尺寸校验报告.csv"
    fieldnames = [
        "源文件", "PDF文件", "输出图像", "源图模式", "DPI",
        "原图宽", "原图高",
        "PDF页面宽(pt)", "PDF页面高(pt)",
        "渲染宽", "渲染高", "最终宽", "最终高",
        "尺寸一致", "状态", "错误信息",
    ]

    with open(report_path, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(report_rows)

    print(
        f"\n目录完成：{input_root}\n"
        f"成功：{success_count} | 失败：{failed_count} | "
        f"重复跳过：{skipped_count}\n"
        f"报告：{report_path}"
    )

    return success_count, failed_count, skipped_count


class QueueWindow:
    def __init__(self, root):
        self.root = root
        self.root.title("图像 → PDF → 图像｜Mask 坐标保护｜多目录排队")
        self.root.geometry("800x520")
        self.root.minsize(680, 440)

        self.directories: list[Path] = []

        tk.Label(
            root,
            text="添加多个目录，按列表顺序依次处理",
            font=("Microsoft YaHei UI", 12, "bold"),
        ).pack(anchor="w", padx=14, pady=(14, 6))

        tk.Label(
            root,
            text=(
                f"DPI={DPI}｜输出图片追加 _p2i｜严格尺寸校验｜"
                "禁止插值缩放、旋转、裁切和补边"
            ),
            fg="#555555",
            anchor="w",
        ).pack(fill="x", padx=14, pady=(0, 8))

        self.listbox = tk.Listbox(
            root,
            selectmode=tk.EXTENDED,
            font=("Consolas", 10),
            activestyle="dotbox",
        )
        self.listbox.pack(fill="both", expand=True, padx=14, pady=6)

        button_frame = tk.Frame(root)
        button_frame.pack(fill="x", padx=14, pady=8)

        tk.Button(
            button_frame,
            text="添加目录",
            width=13,
            command=self.add_directory,
        ).pack(side="left", padx=(0, 6))

        tk.Button(
            button_frame,
            text="移除选中",
            width=13,
            command=self.remove_selected,
        ).pack(side="left", padx=6)

        tk.Button(
            button_frame,
            text="清空队列",
            width=13,
            command=self.clear_queue,
        ).pack(side="left", padx=6)

        tk.Button(
            button_frame,
            text="开始排队处理",
            width=16,
            bg="#1677d2",
            fg="white",
            command=self.start,
        ).pack(side="right")

        self.status = tk.Label(
            root,
            text="当前队列：0 个目录",
            anchor="w",
            fg="#444444",
        )
        self.status.pack(fill="x", padx=14, pady=(0, 12))

    def refresh(self):
        self.listbox.delete(0, tk.END)
        for index, directory in enumerate(self.directories, 1):
            self.listbox.insert(tk.END, f"{index:02d}. {directory}")
        self.status.config(text=f"当前队列：{len(self.directories)} 个目录")

    def add_directory(self):
        selected = filedialog.askdirectory(
            parent=self.root,
            title="选择要加入处理队列的目录",
            mustexist=True,
        )
        if not selected:
            return

        path = Path(selected).resolve()
        existing = {str(p).casefold() for p in self.directories}

        if str(path).casefold() in existing:
            messagebox.showinfo(
                "提示",
                "这个目录已经在队列中了。",
                parent=self.root,
            )
            return

        self.directories.append(path)
        self.refresh()

    def remove_selected(self):
        selected_indices = list(self.listbox.curselection())
        if not selected_indices:
            return

        for index in reversed(selected_indices):
            del self.directories[index]
        self.refresh()

    def clear_queue(self):
        self.directories.clear()
        self.refresh()

    def start(self):
        if not self.directories:
            messagebox.showwarning(
                "提示",
                "请先添加至少一个目录。",
                parent=self.root,
            )
            return

        queue = list(self.directories)
        preview = "\n".join(
            f"{i}. {p}" for i, p in enumerate(queue, 1)
        )

        confirmed = messagebox.askyesno(
            "确认开始",
            f"即将按顺序处理 {len(queue)} 个目录：\n\n"
            f"{preview}\n\n"
            f"DPI：{DPI}\n\n"
            "尺寸异常时将直接报错，不会缩放或裁切修正。\n\n"
            "确定开始吗？",
            parent=self.root,
        )
        if not confirmed:
            return

        self.root.destroy()
        run_queue(queue)


def run_queue(queue: list[Path]):
    globally_processed: set[str] = set()
    total_success = 0
    total_failed = 0
    total_skipped = 0

    for queue_index, directory in enumerate(queue, 1):
        print(f"\n\n######## 队列 {queue_index}/{len(queue)} ########")
        try:
            success, failed, skipped = process_directory(
                directory,
                globally_processed,
            )
            total_success += success
            total_failed += failed
            total_skipped += skipped
        except Exception:
            total_failed += 1
            traceback.print_exc()

    summary = (
        "全部队列处理完成！\n\n"
        f"目录数量：{len(queue)}\n"
        f"成功图片：{total_success}\n"
        f"失败图片：{total_failed}\n"
        f"重复跳过：{total_skipped}\n\n"
        "每个目录的详细结果请查看：\n"
        "对应的 *_PDF2IMG\\尺寸校验报告.csv\n\n"
        "注意：失败图片不会自动缩放、裁切或补边。"
    )

    print("\n" + "=" * 90 + "\n" + summary)
    root = tk.Tk()
    root.withdraw()
    root.attributes("-topmost", True)
    messagebox.showinfo("处理完成", summary, parent=root)
    root.destroy()


def main():
    root = tk.Tk()
    QueueWindow(root)
    root.mainloop()


if __name__ == "__main__":
    try:
        main()
    except Exception:
        traceback.print_exc()
        try:
            error_root = tk.Tk()
            error_root.withdraw()
            messagebox.showerror(
                "程序异常",
                traceback.format_exc(),
                parent=error_root,
            )
            error_root.destroy()
        except Exception:
            pass
        sys.exit(1)

后续验收 / 改进项

• 使用 512×512、奇数尺寸、不同长宽比的真实图像检查最终像素宽高和 Mask 坐标。
• 用 RGB、RGBA、灰度图分别验证 PDF 往返；检查图像模式、透明度变化的可接受性。
• 确认 16-bit/浮点 Mask 被明确拒绝，不产生错误转换结果。
• 确认多个输入目录包含/重叠时不重复处理，输出子目录结构与报告路径正确。
• 测试同目录下同名不同扩展名（如 a.jpg 和 a.png）的 PDF/PNG 输出冲突，必要时增加防覆盖策略。
• 在实际 Windows 环境与目标 PyMuPDF 版本下运行回归测试（此 Issue 仅归档源码，尚未执行环境验证）。

关联：边缘模型部署时，同样要确保预处理 resize/letterbox/padding 与坐标还原一致。