Python 实战：从零开发一个 PDF 发票管理系统，支持重命名、金额统计、发票查重

摘要

本文基于 Python 和 PyMuPDF，完整实现了一套 PDF 发票管理系统，支持发票重命名、金额统计和发票号码查重。项目从真实需求出发，针对发票文件名混乱、手工统计低效、重复票据难排查等问题，设计了统一的核心解析模块和命令行交互入口。系统可自动提取发票中的项目名称、价税合计和发票号码，并完成批量处理与日志记录。文章不仅讲解了功能实现过程，还对项目结构、技术选型、重构思路和后续扩展方向进行了系统整理，适合作为 Python 自动化实战案例，也适合个人、团队和实验室直接复用。

---

一、项目背景

日常做报销、采购归档、团队物资管理时，最头疼的事情之一就是发票整理。

常见问题基本都差不多：

• 发票 PDF 文件名混乱，根本看不出内容
• 发票数量一多，手工统计金额非常慢
• 同一张发票可能被重复保存，人工查重费时费力

如果只是处理几张票，手工当然也能完成；但一旦数量上来，重复劳动就会非常明显。

所以我做了一个基于 Python 的 PDF 发票管理系统，专门解决下面三件事：

1. 自动读取发票 PDF 内容
2. 批量重命名发票文件
3. 自动统计金额并检查重复发票

这篇文章不是单纯讲“怎么运行脚本”，而是直接给出一篇可复现、可二次开发、可直接发布的完整项目说明，并把核心源码一并放进文章里。

---

二、最终实现的功能

这套系统目前保留了三个核心功能。

1. 发票重命名

自动从 PDF 中提取“项目名称”和“金额”，将文件重命名为：

项目名称_金额元.pdf

例如：

螺丝刀_28.16元.pdf
3D打印机线材_148.75元.pdf
控制器_989.00元.pdf

2. 发票金额统计

批量识别发票中的“价税合计”，自动累计总金额，并输出需要人工核对的文件。

3. 发票号码查重

自动提取发票号码，识别重复票据，避免重复报销或重复归档。

---

三、技术选型

本项目用到的技术非常简单，但很适合做这种自动化工具：

• Python
• PyMuPDF
• re 正则表达式
• pathlib
• argparse

为什么选择这套方案？

• Python 开发效率高，适合做批量处理工具
• PyMuPDF 对 PDF 文本提取很方便，速度也足够快
• 标准库已经能覆盖文件处理、命令行参数、日志写入等需求
• 整体依赖很少，方便别人复现和部署

项目依赖如下：

PyMuPDF>=1.24,<2.0

安装命令：

python -m pip install PyMuPDF

---

四、开发思路：先拆问题，再做功能

做这类项目时，最重要的不是一开始就堆很多功能，而是先找到最小闭环。

这个发票管理系统的核心其实只有一句话：

从 PDF 中提取结构化信息，再基于这些信息完成批量处理。

只要这一步成立，后面很多功能都能顺着做出来。

因此系统的开发思路可以拆成三层：

第一层：PDF 文本提取

先把发票里的文字提出来。

第二层：关键信息解析

从文本中提取：

• 项目名称
• 金额
• 发票号码

第三层：业务操作

基于这些字段完成：

• 文件重命名
• 金额统计
• 重复检测

这套思路不仅适用于发票，换成合同、成绩单、报告、证书等 PDF 文档，同样成立。

---

五、项目结构设计

为了避免后期功能越来越乱，建议把项目整理成统一结构：

invoice_project/
├─ invoice_system/
│  ├─ __init__.py
│  ├─ cli.py
│  └─ core.py
├─ main.py
├─ invoice_manager.py
├─ rename_short.py
├─ total_money.py
├─ check_duplicate.py
└─ requirements.txt

各文件职责如下：

• invoice_system/core.py
  负责核心业务逻辑，包括文本提取、字段解析、重命名、金额统计、查重。

• invoice_system/cli.py
  负责命令行参数、菜单交互和输出展示。

• main.py
  统一主入口。

• invoice_manager.py

• rename_short.py

• total_money.py

• check_duplicate.py

  这些旧脚本可以保留下来作为兼容入口，但底层统一调用同一套核心逻辑，不再各自维护重复代码。

这样做的好处很明显：

• 后续升级只需要改核心模块
• 不会再出现多个脚本各写一套正则的情况
• 老用户仍然可以按原来的方式运行程序

---

六、核心难点：如何从 PDF 中提取发票信息

这个项目最关键的一步，是把 PDF 里的文本提取出来。

基本写法如下：

with fitz.open(pdf_path) as doc:
    text = "".join(page.get_text("text", sort=True) for page in doc)

拿到整份文本后，再通过正则表达式识别关键字段。

1. 提取金额

金额对应的是发票里的“价税合计（小写）”。

MONEY_RE = re.compile(r"价税合计.*?小写.*?[¥￥]\s*([\d,]+\.\d{1,2})", re.S)

2. 提取项目名称

项目名称通常位于“项目名称”字段附近，所以可以匹配这个标签后的首个有效文本：

DEVICE_RE = re.compile(r"项目名称.*?\n(?:\s*\n)*([^\n]+)", re.S)

3. 提取发票号码

INVOICE_NO_RE = re.compile(r"(?:发票号码|发票代码|号码)[:：]?\s*(\d{6,})", re.S)

这种写法的核心目的不是追求“完美匹配”，而是尽量兼容更多电子发票模板。

---

七、完整源码

下面直接给出一套可以落地使用的完整源码。为了方便读者复制，代码按文件拆分展示。

1. invoice_system/core.py

from __future__ import annotations

import re
from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path

import fitz

MONEY_RE = re.compile(r"价税合计.*?小写.*?[¥￥]\s*([\d,]+\.\d{1,2})", re.S)
DEVICE_RE = re.compile(r"项目名称.*?\n(?:\s*\n)*([^\n]+)", re.S)
INVOICE_NO_RE = re.compile(r"(?:发票号码|发票代码|号码)[:：]?\s*(\d{6,})", re.S)


@dataclass
class InvoiceRecord:
    path: Path
    amount: float | None
    item_name: str | None
    invoice_no: str | None
    error: str | None = None


@dataclass
class RenameResult:
    source_name: str
    target_name: str | None
    status: str
    message: str


@dataclass
class AmountResult:
    file_name: str
    amount: float | None
    status: str
    message: str


@dataclass
class DuplicateResult:
    invoice_no: str
    file_names: list[str]


def ensure_logs_dir(base_dir: Path) -> Path:
    logs_dir = base_dir / "logs"
    logs_dir.mkdir(parents=True, exist_ok=True)
    return logs_dir


def write_log(log_path: Path, message: str, *, overwrite: bool = False) -> None:
    log_path.parent.mkdir(parents=True, exist_ok=True)
    mode = "w" if overwrite else "a"
    with log_path.open(mode, encoding="utf-8") as file:
        file.write(f"[{datetime.now():%Y-%m-%d %H:%M:%S}] {message}\n")


def normalize_folder(folder: str | Path) -> Path:
    path = Path(str(folder).strip().strip('"')).expanduser()
    return path.resolve()


def list_pdf_files(folder: Path) -> list[Path]:
    return sorted(p for p in folder.glob("*.pdf") if p.is_file())


def extract_text(pdf_path: Path) -> str:
    with fitz.open(pdf_path) as doc:
        return "".join(page.get_text("text", sort=True) for page in doc)


def sanitize_item_name(raw_name: str) -> str:
    cleaned = re.sub(r"[^\u4e00-\u9fa5a-zA-Z0-9]", "", raw_name.strip())
    return cleaned[:24]


def parse_invoice(pdf_path: Path) -> InvoiceRecord:
    try:
        text = extract_text(pdf_path)
        money_match = MONEY_RE.search(text)
        device_match = DEVICE_RE.search(text)
        invoice_match = INVOICE_NO_RE.search(text)

        amount = None
        item_name = None
        invoice_no = None

        if money_match:
            amount = float(money_match.group(1).replace(",", ""))
        if device_match:
            item_name = sanitize_item_name(device_match.group(1).split()[0])
        if invoice_match:
            invoice_no = invoice_match.group(1)

        return InvoiceRecord(
            path=pdf_path,
            amount=amount,
            item_name=item_name or None,
            invoice_no=invoice_no or None,
        )
    except Exception as exc:
        return InvoiceRecord(
            path=pdf_path,
            amount=None,
            item_name=None,
            invoice_no=None,
            error=str(exc),
        )


def unique_name(folder: Path, file_name: str) -> str:
    candidate = folder / file_name
    stem = candidate.stem
    suffix = candidate.suffix
    counter = 1
    while candidate.exists():
        candidate = folder / f"{stem}_{counter}{suffix}"
        counter += 1
    return candidate.name


def rename_invoices(folder: Path, project_root: Path) -> list[RenameResult]:
    pdf_files = list_pdf_files(folder)
    log_path = ensure_logs_dir(project_root) / "invoice_rename.log"
    write_log(log_path, f"========== 重命名开始 {len(pdf_files)} 张 ==========", overwrite=True)

    results: list[RenameResult] = []
    for pdf_file in pdf_files:
        record = parse_invoice(pdf_file)
        if record.error:
            msg = f"读取失败: {record.error}"
            results.append(RenameResult(pdf_file.name, None, "error", msg))
            write_log(log_path, f"[ERROR] {pdf_file.name} {msg}")
            continue

        if record.amount is None or not record.item_name:
            msg = "缺少项目名称或金额"
            results.append(RenameResult(pdf_file.name, None, "skipped", msg))
            write_log(log_path, f"[SKIP] {pdf_file.name} {msg}")
            continue

        target_name = unique_name(folder, f"{record.item_name}_{record.amount:.2f}元.pdf")
        target_path = folder / target_name
        try:
            pdf_file.rename(target_path)
            results.append(RenameResult(pdf_file.name, target_name, "ok", "重命名成功"))
            write_log(log_path, f"[OK] {pdf_file.name} -> {target_name}")
        except Exception as exc:
            msg = f"重命名失败: {exc}"
            results.append(RenameResult(pdf_file.name, None, "error", msg))
            write_log(log_path, f"[ERROR] {pdf_file.name} {msg}")

    ok_count = sum(1 for item in results if item.status == "ok")
    write_log(log_path, f"========== 重命名完成 {ok_count}/{len(pdf_files)} 张 ==========")
    return results


def summarize_amounts(folder: Path, project_root: Path) -> tuple[float, list[AmountResult]]:
    pdf_files = list_pdf_files(folder)
    log_path = ensure_logs_dir(project_root) / "amount_summary.log"
    write_log(log_path, f"========== 金额统计开始 {len(pdf_files)} 张 ==========", overwrite=True)

    total = 0.0
    results: list[AmountResult] = []
    for pdf_file in pdf_files:
        record = parse_invoice(pdf_file)
        if record.error:
            msg = f"读取失败: {record.error}"
            results.append(AmountResult(pdf_file.name, None, "error", msg))
            write_log(log_path, f"[ERROR] {pdf_file.name} {msg}")
            continue

        if record.amount is None:
            msg = "未识别到金额"
            results.append(AmountResult(pdf_file.name, None, "skipped", msg))
            write_log(log_path, f"[SKIP] {pdf_file.name} {msg}")
            continue

        total += record.amount
        results.append(AmountResult(pdf_file.name, record.amount, "ok", "识别成功"))
        write_log(log_path, f"[OK] {pdf_file.name} -> {record.amount:.2f}")

    failed_count = sum(1 for item in results if item.status != "ok")
    write_log(log_path, f"========== 金额统计完成 总金额={total:.2f} 需核对={failed_count} 张 ==========")
    return total, results


def find_duplicates(folder: Path, project_root: Path) -> tuple[list[DuplicateResult], list[str]]:
    pdf_files = list_pdf_files(folder)
    log_path = ensure_logs_dir(project_root) / "duplicate_check.log"
    write_log(log_path, f"========== 发票查重开始 {len(pdf_files)} 张 ==========", overwrite=True)

    invoice_map: defaultdict[str, list[str]] = defaultdict(list)
    missing_invoice_no: list[str] = []

    for pdf_file in pdf_files:
        record = parse_invoice(pdf_file)
        if record.error:
            missing_invoice_no.append(pdf_file.name)
            write_log(log_path, f"[ERROR] {pdf_file.name} 读取失败: {record.error}")
            continue

        if not record.invoice_no:
            missing_invoice_no.append(pdf_file.name)
            write_log(log_path, f"[NO_INV] {pdf_file.name} 未识别到发票号码")
            continue

        invoice_map[record.invoice_no].append(pdf_file.name)

    duplicates = [
        DuplicateResult(invoice_no=invoice_no, file_names=file_names)
        for invoice_no, file_names in sorted(invoice_map.items())
        if len(file_names) > 1
    ]

    if duplicates:
        write_log(log_path, f"========== 发现 {len(duplicates)} 组重复发票 ==========")
        for item in duplicates:
            write_log(log_path, f"[DUPLICATE] {item.invoice_no} -> {', '.join(item.file_names)}")
    else:
        write_log(log_path, "========== 未发现重复发票 ==========")

    write_log(log_path, f"========== 发票查重完成 无号码={len(missing_invoice_no)} 张 ==========")
    return duplicates, missing_invoice_no

2. invoice_system/cli.py

from __future__ import annotations

import argparse
from pathlib import Path

from .core import find_duplicates
from .core import list_pdf_files
from .core import normalize_folder
from .core import rename_invoices
from .core import summarize_amounts


def _prompt_folder() -> Path:
    folder = input("请把发票 pdf 文件夹拖进来，然后按回车：")
    return normalize_folder(folder)


def _resolve_folder(folder_arg: str | None) -> Path:
    return normalize_folder(folder_arg) if folder_arg else _prompt_folder()


def _print_rename(folder: Path, project_root: Path) -> None:
    pdf_files = list_pdf_files(folder)
    if not pdf_files:
        print("目录里没有找到 pdf。")
        return

    results = rename_invoices(folder, project_root)
    for item in results:
        if item.status == "ok":
            print(f"OK   {item.source_name} -> {item.target_name}")
        elif item.status == "skipped":
            print(f"SKIP {item.source_name} -> {item.message}")
        else:
            print(f"ERR  {item.source_name} -> {item.message}")

    ok_count = sum(1 for item in results if item.status == "ok")
    print("----------")
    print(f"重命名完成：{ok_count}/{len(results)} 张")
    print(f"日志位置：{project_root / 'logs' / 'invoice_rename.log'}")


def _print_summary(folder: Path, project_root: Path) -> None:
    pdf_files = list_pdf_files(folder)
    if not pdf_files:
        print("目录里没有找到 pdf。")
        return

    total, results = summarize_amounts(folder, project_root)
    for item in results:
        if item.status == "ok":
            print(f"OK   {item.file_name} -> {item.amount:.2f}")
        elif item.status == "skipped":
            print(f"SKIP {item.file_name} -> {item.message}")
        else:
            print(f"ERR  {item.file_name} -> {item.message}")

    need_check = sum(1 for item in results if item.status != "ok")
    print("----------")
    print(f"成功识别金额合计：{total:.2f}")
    print(f"需要人工核对：{need_check} 张")
    print(f"日志位置：{project_root / 'logs' / 'amount_summary.log'}")


def _print_duplicates(folder: Path, project_root: Path) -> None:
    pdf_files = list_pdf_files(folder)
    if not pdf_files:
        print("目录里没有找到 pdf。")
        return

    duplicates, missing = find_duplicates(folder, project_root)
    print("----------")
    if duplicates:
        print(f"发现 {len(duplicates)} 组重复发票号码：")
        for item in duplicates:
            print(f"号码 {item.invoice_no}:")
            for file_name in item.file_names:
                print(f"  - {file_name}")
    else:
        print("未发现重复发票号码。")

    if missing:
        print(f"未识别到发票号码：{len(missing)} 张")
    print(f"日志位置：{project_root / 'logs' / 'duplicate_check.log'}")


def run_action(action: str, folder_arg: str | None = None) -> None:
    project_root = Path(__file__).resolve().parent.parent
    folder = _resolve_folder(folder_arg)

    if not folder.exists() or not folder.is_dir():
        print(f"目录不存在：{folder}")
        return

    if action == "rename":
        _print_rename(folder, project_root)
    elif action == "summary":
        _print_summary(folder, project_root)
    elif action == "duplicate":
        _print_duplicates(folder, project_root)
    else:
        print(f"未知操作：{action}")


def _menu() -> str:
    print("=" * 52)
    print("发票管理系统")
    print("=" * 52)
    print("1. 发票重命名")
    print("2. 发票金额统计")
    print("3. 发票号码查重")
    print("4. 退出")
    return input("请输入选择 (1/2/3/4): ").strip()


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="发票管理系统")
    parser.add_argument(
        "--action",
        choices=["rename", "summary", "duplicate", "menu"],
        default="menu",
        help="指定执行动作，不传时进入交互菜单",
    )
    parser.add_argument("--folder", help="待处理 PDF 所在目录")
    return parser


def main() -> None:
    parser = build_parser()
    args = parser.parse_args()

    if args.action != "menu":
        run_action(args.action, args.folder)
        return

    choice = _menu()
    if choice == "1":
        run_action("rename")
    elif choice == "2":
        run_action("summary")
    elif choice == "3":
        run_action("duplicate")
    elif choice == "4":
        print("程序已退出。")
    else:
        print("无效选择，请重新运行程序。")


if __name__ == "__main__":
    main()

3. main.py

from invoice_system.cli import main


if __name__ == "__main__":
    main()

4. rename_short.py

import argparse

from invoice_system.cli import run_action


def main() -> None:
    parser = argparse.ArgumentParser(description="发票重命名")
    parser.add_argument("--folder", help="待处理 PDF 所在目录")
    args = parser.parse_args()
    run_action("rename", args.folder)


if __name__ == "__main__":
    main()

5. total_money.py

import argparse

from invoice_system.cli import run_action


def main() -> None:
    parser = argparse.ArgumentParser(description="发票金额统计")
    parser.add_argument("--folder", help="待处理 PDF 所在目录")
    args = parser.parse_args()
    run_action("summary", args.folder)


if __name__ == "__main__":
    main()

6. check_duplicate.py

import argparse

from invoice_system.cli import run_action


def main() -> None:
    parser = argparse.ArgumentParser(description="发票号码查重")
    parser.add_argument("--folder", help="待处理 PDF 所在目录")
    args = parser.parse_args()
    run_action("duplicate", args.folder)


if __name__ == "__main__":
    main()

7. invoice_manager.py

from invoice_system.cli import main


if __name__ == "__main__":
    main()

8. requirements.txt

PyMuPDF>=1.24,<2.0

---

八、功能实现详解

1. 发票重命名

重命名功能的流程如下：

1. 遍历指定目录下的所有 PDF
2. 读取每个 PDF 的文本
3. 解析项目名称和金额
4. 生成新文件名
5. 处理同名冲突
6. 执行重命名并记录日志

生成文件名时，采用的格式是：

项目名称_金额元.pdf

如果遇到重名文件，就自动追加编号：

螺丝刀_28.16元.pdf
螺丝刀_28.16元_1.pdf
螺丝刀_28.16元_2.pdf

这种方式的优点是：

• 文件名一眼可读
• 目录排序清晰
• 不会覆盖已有文件

2. 金额统计

金额统计功能的流程也很直接：

1. 遍历全部 PDF
2. 提取每张发票的金额
3. 识别成功就累加
4. 未识别成功的文件加入人工核对列表
5. 输出总金额和日志

这种做法比纯粹输出总金额更可靠，因为它保留了“失败清单”，便于后续核对。

3. 发票查重

查重的逻辑是：

1. 提取每个 PDF 的发票号码
2. 用字典按发票号码归组
3. 出现次数大于 1 的号码，判定为重复

核心思路很像这样：

invoice_map[invoice_no].append(file_name)

最后把长度大于 1 的分组筛出来即可。

这个方法足够轻量，而且对于多数实际场景已经够用。

---

九、为什么项目后期一定要重构整理

很多人做自动化脚本时，早期都会经历一个阶段：

• 先写一个重命名脚本
• 再补一个统计脚本
• 再补一个查重脚本
• 最后再写一个菜单脚本把它们串起来

这在前期完全合理，因为目标是先把事情做成。

但如果一直不整理，后面会出现几个非常典型的问题：

• 同样的正则表达式在多个文件里重复出现
• 改一个 bug 要同步改多个脚本
• 日志文件散落在根目录
• 新用户根本不知道该运行哪个文件

所以项目做到后面，必须做一次真正的“工程化整理”。

我这次做的整理主要包括：

1. 把通用逻辑统一收敛到 core.py
2. 把交互与参数统一放进 cli.py
3. 增加 main.py 作为统一主入口
4. 保留旧脚本作为兼容入口
5. 增加 requirements.txt
6. 把日志统一收纳到 logs/

这一步非常重要，因为它决定了你的项目是“自己能跑的小脚本”，还是“别人也能接手的完整项目”。

---

十、项目运行方式

1. 创建虚拟环境

python -m venv venv

2. 激活环境

.\venv\Scripts\Activate.ps1

3. 安装依赖

python -m pip install -r .\requirements.txt

4. 运行主程序

python .\main.py

运行后会看到菜单：

====================================================
发票管理系统
====================================================
1. 发票重命名
2. 发票金额统计
3. 发票号码查重
4. 退出

选择功能后，把发票所在文件夹拖到终端窗口，回车即可。

---

十一、也支持命令行直接执行

如果不想走交互菜单，也可以直接通过命令行参数指定功能。

1. 重命名

python .\main.py --action rename --folder "D:\发票目录"

2. 金额统计

python .\main.py --action summary --folder "D:\发票目录"

3. 发票查重

python .\main.py --action duplicate --folder "D:\发票目录"

---

十二、兼容旧脚本入口

为了兼容以前的使用方式，这几个脚本也仍然可以单独运行：

python .\invoice_manager.py
python .\rename_short.py
python .\total_money.py
python .\check_duplicate.py

如果需要直接指定目录，也支持：

python .\rename_short.py --folder "D:\发票目录"
python .\total_money.py --folder "D:\发票目录"
python .\check_duplicate.py --folder "D:\发票目录"

---

十三、日志输出说明

为了便于定位问题，程序运行时会自动写日志。

日志统一保存在 logs/ 目录下：

• logs/invoice_rename.log
• logs/amount_summary.log
• logs/duplicate_check.log

日志的作用主要有两个：

1. 记录每次运行的处理结果
2. 方便排查哪些文件解析失败、哪些文件存在重复

对于批量处理类项目来说，日志几乎是必备项。没有日志，程序一旦跑出异常，后续排查会非常麻烦。

---

十四、这套思路为什么有普遍性

这篇文章虽然讲的是发票管理，但本质上解决的是一个更通用的问题：

如何把一批非结构化 PDF 文档，自动转换成结构化信息，再驱动批量处理。

同样的方法可以迁移到很多场景：

• 合同批量重命名
• 证书信息提取
• 成绩单归档
• 论文 PDF 分类整理
• 采购单、出库单、报销单自动归档

所以这个项目真正值得复用的，不只是代码，而是一整套开发方法。

---

十五、后续可以继续扩展的方向

如果要把这个项目继续升级，还可以做很多增强：

1. 导出 Excel 汇总表

把发票名称、金额、发票号码、状态导出成 Excel，形成完整台账。

2. 增加 GUI 图形界面

让不懂命令行的用户也能直接点击操作。

3. 接入 OCR

目前系统适合可提取文本的电子 PDF。如果是扫描版发票，可以加 OCR 支持。

4. 增加更多字段提取

例如：

• 开票日期
• 销售方名称
• 税额
• 购买方名称

5. 自动分类归档

按月份、项目、类别自动整理到不同文件夹中。

如果这些功能继续补上，这个项目就不再只是一个小工具，而会逐渐变成一套更完整的文档管理应用。

---

十六、总结

这套发票管理系统，从开发思路上可以概括为五步：

1. 从真实痛点出发
2. 先完成最小可用闭环
3. 再逐步补充统计和查重功能
4. 最后做统一重构和项目整理
5. 补齐文档，让别人也能直接复现

对于很多 Python 自动化项目来说，这其实是一个非常通用的路线：

• 前期先解决问题
• 中期逐步丰富功能
• 后期再做结构化和工程化整理

如果你也在做类似的 PDF 自动化处理项目，希望这篇文章能给你一个完整、可落地的参考。

---

十七、结尾

如果你后面准备继续把这套系统升级成：

• 支持 Excel 导出
• 支持 OCR
• 支持图形界面
• 支持数据库归档

那这版结构已经足够作为后续迭代的基础版本。

这类项目最重要的，不是代码炫技，而是真能解决实际问题。能把一个原本靠手工完成、又重复又容易出错的流程自动化，本身就已经很有价值。