夜雨聆风PDF 发票批量识别与归档——财务人的电子发票管理神器
随着全面数字化电子发票的普及，财务人面临一个新挑战：
每月收到几百张 PDF 格式的电子发票，需要逐张打开查看金额、核对信息、分类归档……
如果手动操作，这是一项极其耗时且容易出错的工作。而用 Python 实现自动化后：
• ✅ 批量读取 PDF 发票中的关键信息 • ✅ 自动提取 发票号码、金额、日期、开票方 • ✅ 智能分类 按供应商/费用类型归档 • ✅ 汇总统计 自动生成发票台账 • ✅ 重复检测 防止同一张发票报销两次 
一、 工具准备
安装必要的库
# PDF 文本提取
pip install PyPDF2 pdfplumber
# PDF 表格提取（更精确）
pip install tabula-py
# OCR（如果 PDF 是扫描件图片）
# pip install pytesseract pdf2image
# 还需要安装 Tesseract OCR 引擎库的选择指南
PyPDF2 | ||
pdfplumber | ||
tabula-py | ||
pdf2image + pytesseract |
二、 基础操作：读取 PDF 内容
import pdfplumber
import pandas as pd
import os
import re
from datetime import datetime
defextract_text_from_pdf(pdf_path):
"""
从 PDF 文件中提取所有文本内容
"""
text_content = []
try:
with pdfplumber.open(pdf_path) as pdf:
print(f"📄 文件: {os.path.basename(pdf_path)}")
print(f" 页数: {len(pdf.pages)}")
for i, page inenumerate(pdf.pages):
text = page.extract_text()
if text:
text_content.append({
"页码": i + 1,
"内容": text
})
# 打印前200字符预览
if i == 0:
print(f" 第1页预览: {text[:200]}...")
except Exception as e:
print(f" ❌ 读取失败: {e}")
returnNone
return text_content
# 测试
# extract_text_from_pdf("示例发票.pdf")三、 核心功能：发票信息智能提取
电子发票的关键字段识别
中国电子发票（增值税普通发票/专用发票）通常包含以下关键字段：
classInvoiceExtractor:
"""电子发票信息提取器"""
# 关键字正则模式
PATTERNS = {
"发票号码": r"发票号码[：:]\s*(\d{8,12})",
"发票代码": r"发票代码[：:]\s*(\d{10,12})",
"开票日期": r"开票日期[：:]\s*(\d{4}年\d{1,2}月\d{1,2}日|\d{4}-\d{1,2}-\d{1,2})",
"校验码": r"校验码[：:]\s*(\w+)",
"购买方名称": r"购买方[名称]*[：:]\s*(.+?)(?:\n|纳税人)",
"销售方名称": r"销售方[名称]*[：:]\s*(.+?)(?:\n|纳税人)",
"价税合计(大写)": r"价税合计.*?大写[）:].*?(\S+圆整)",
"价税合计(小写)": r"价税合计.*?小写[）:]\s*[¥￥]?\s*(\d+\.?\d*)",
"金额合计": r"合计.*?金额[）:]\s*[¥￥]?\s*(\d+\.?\d*)",
"税额合计": r"合计.*?税额[）:]\s*[¥￥]?\s*(\d+\.?\d*)",
}
def__init__(self):
self.extracted_data = []
defextract_invoice_info(self, pdf_path):
"""
从单个 PDF 发票中提取结构化信息
"""
result = {
"文件名": os.path.basename(pdf_path),
"文件路径": pdf_path,
"提取时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
try:
with pdfplumber.open(pdf_path) as pdf:
# 合并所有页面文本
full_text = ""
for page in pdf.pages:
text = page.extract_text()
if text:
full_text += text + "\n"
result["原始文本"] = full_text[:500] # 保存前500字符用于调试
# 使用正则表达式提取各字段
for field_name, pattern inself.PATTERNS.items():
match = re.search(pattern, full_text)
ifmatch:
result[field_name] = match.group(1).strip()
else:
result[field_name] = None
# 尝试提取表格数据（明细行）
tables = []
for page in pdf.pages:
tables.extend(page.extract_tables() or [])
if tables:
# 通常第一个表格是发票明细
result["表格数据"] = tables[0]
result["表格行数"] = len(tables[0])
except Exception as e:
result["错误"] = str(e)
self.extracted_data.append(result)
return result
defformat_result(self, result):
"""格式化输出提取结果"""
print("\n" + "=" * 50)
print(f"📋 发票信息: {result['文件名']}")
print("=" * 50)
display_fields = [
"发票代码", "发票号码", "开票日期",
"购买方名称", "销售方名称",
"价税合计(小写)", "金额合计", "税额合计"
]
for field in display_fields:
value = result.get(field)
if value:
print(f" {field}: {value}")
else:
print(f" {field}: (未识别)")
if"错误"in result:
print(f" ⚠️ 错误: {result['错误']}")
return result四、 批量处理：整个文件夹的发票
defbatch_process_invoices(folder_path, output_file=None):
"""
批量处理文件夹中的所有 PDF 发票
"""
print("=" * 60)
print("📦 PDF 发票批量处理系统")
print("=" * 60)
# 1. 扫描文件夹中的 PDF 文件
pdf_files = []
for root, dirs, files in os.walk(folder_path):
for file in files:
if file.lower().endswith(".pdf"):
pdf_files.append(os.path.join(root, file))
ifnot pdf_files:
print(f"❌ 在 '{folder_path}' 中未找到 PDF 文件")
returnNone
print(f"\n📂 找到 {len(pdf_files)} 个 PDF 文件")
# 2. 逐个提取
extractor = InvoiceExtractor()
results = []
for i, pdf_path inenumerate(pdf_files, 1):
print(f"\n[{i}/{len(pdf_files)}] 处理中...")
result = extractor.extract_invoice_info(pdf_path)
extractor.format_result(result)
results.append(result)
# 3. 汇总为 DataFrame
summary_data = []
for r in results:
row = {
"文件名": r.get("文件名"),
"发票代码": r.get("发票代码"),
"发票号码": r.get("发票号码"),
"开票日期": r.get("开票日期"),
"购买方": r.get("购买方名称"),
"销售方": r.get("销售方名称"),
"金额(元)": r.get("金额合计"),
"税额(元)": r.get("税额合计"),
"价税合计(元)": r.get("价税合计(小写)"),
"提取状态": "✅ 成功"if r.get("发票号码") else"⚠️ 不完整",
"提取时间": r.get("提取时间")
}
summary_data.append(row)
df = pd.DataFrame(summary_data)
# 4. 统计摘要
print(f"\n{'='*60}")
print(f"📊 处理结果统计:")
print(f" 总文件数: {len(df)}")
print(f" 成功提取: {len(df[df['提取状态']=='✅ 成功'])}")
print(f" 提取不完整: {len(df[df['提取状态']!='✅ 成功'])}")
# 金额汇总（如果能提取到的话）
amount_col = "价税合计(元)"
valid_amounts = pd.to_numeric(df[amount_col], errors="coerce").dropna()
iflen(valid_amounts) > 0:
print(f" 金额合计: {valid_amounts.sum():,.2f} 元")
print(f" 平均金额: {valid_amounts.mean():,.2f} 元")
# 5. 保存结果
if output_file isNone:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
output_file = f"发票台账_{timestamp}.xlsx"
df.to_excel(output_file, index=False)
print(f"\n💾 台账已保存: {output_file}")
# 6. 重复检测
duplicate_numbers = df[df["发票号码"].notna() & df["发票号码"].duplicated(keep=False)]
iflen(duplicate_numbers) > 0:
print(f"\n⚠️ 发现 {len(duplicate_numbers)} 张疑似重复发票:")
print(duplicate_numbers[["文件名", "发票号码"]].to_string(index=False))
return df
# ===== 使用示例 =====
# batch_process_invoices(
# folder_path="./待处理发票/",
# output_file="发票台账_2024Q1.xlsx"
# )五、 高级功能：自动分类归档
defauto_classify_and_archive(invoices_df, source_folder, archive_folder):
"""
根据发票信息自动分类并归档文件
:param invoices_df: 发票台账 DataFrame
:param source_folder: 发票源文件夹
:param archive_folder: 归档目标根目录
"""
import shutil
os.makedirs(archive_folder, exist_ok=True)
# 定义分类规则
classification_rules = [
{"关键词": ["差旅", "火车", "飞机", "酒店"], "类别": "01-差旅费"},
{"关键词": ["餐饮", "食品", "餐饮服务"], "类别": "02-业务招待费"},
{"关键词": ["办公", "耗材", "文具", "打印"], "类别": "03-办公费"},
{"关键词": ["运输", "快递", "物流", "运费"], "类别": "04-运杂费"},
{"关键词": ["广告", "宣传", "推广", "传媒"], "类别": "05-广告费"},
{"关键词": ["咨询", "服务", "技术服务"], "类别": "06-咨询服务费"},
{"关键词": ["租赁", "房租", "物业"], "类别": "07-租赁费"},
]
archive_log = []
for _, invoice in invoices_df.iterrows():
filename = invoice["文件名"]
source_path = os.path.join(source_folder, filename)
ifnot os.path.exists(source_path):
continue
# 匹配分类规则
target_category = "99-其他"
raw_text = invoice.get("原始文本", "") or""
for rule in classification_rules:
ifany(kw in raw_text for kw in rule["关键词"]):
target_category = rule["类别"]
break
# 也可以按销售方名称分类
seller = invoice.get("销售方") or""
if"京东"in seller or"淘宝"in seller:
target_category = "08-电商采购"
# 创建目标目录并复制文件
target_dir = os.path.join(archive_folder, target_category)
os.makedirs(target_dir, exist_ok=True)
target_path = os.path.join(target_dir, filename)
shutil.copy2(source_path, target_path)
archive_log.append({
"原文件": filename,
"目标分类": target_category,
"归档路径": target_path,
"归档时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
})
# 保存归档日志
log_df = pd.DataFrame(archive_log)
log_file = os.path.join(archive_folder, f"归档日志_{datetime.now().strftime('%Y%m%d')}.xlsx")
log_df.to_excel(log_file, index=False)
# 统计各类别数量
print("\n📁 归档统计:")
category_counts = log_df["目标分类"].value_counts()
for cat, count in category_counts.items():
print(f" {cat}: {count} 张")
print(f"\n✅ 归档完成! 共归档 {len(archive_log)} 张发票")
print(f"📁 归档目录: {archive_folder}")
return log_df六、 进阶：OCR 识别扫描件发票
如果你的发票是扫描件或图片格式的 PDF（无法直接提取文字），需要使用 OCR 技术：
defocr_extract_invoice(image_pdf_path):
"""
使用 OCR 从图片型 PDF 中提取文字
注意：需要安装 Tesseract OCR 引擎
"""
try:
import pdf2image
from PIL import Image
import pytesseract
# 将 PDF 转换为图片
images = pdf2image.convert_from_path(image_pdf_path)
all_text = ""
for i, image inenumerate(images):
# OCR 识别（设置中文语言包）
text = pytesseract.image_to_string(
image,
lang="chi_sim+eng", # 中文简体 + 英文
config="--psm 6"# 页面分割模式
)
all_text += f"=== 第{i+1}页 ===\n{text}\n"
print("📝 OCR 识别结果（前500字）:")
print(all_text[:500])
return all_text
except ImportError:
print("❌ 缺少依赖库，请安装:")
print(" pip install pdf2image pytesseract pillow")
print(" 并安装 Tesseract OCR 引擎")
returnNone
except Exception as e:
print(f"❌ OCR 识别失败: {e}")
returnNoneOCR 环境配置提示：
1. 下载安装 Tesseract OCR 2. 下载中文语言包 chi_sim.traineddata3. 设置环境变量 TESSDATA_PREFIX指向语言包目录
七、 完整工作流：从接收到入账
deffull_invoice_workflow(incoming_folder, archive_folder):
"""
发票完整处理工作流：
接收 → 识别 → 分类 → 归档 → 生成台账 → 异常报告
"""
print("=" * 70)
print("🔄 电子发票全自动处理流程")
print("=" * 70)
start_time = datetime.now()
# Step 1: 扫描新发票
print("\n[Step 1/5] 📥 扫描待处理发票...")
new_invoices = batch_process_invoices(incoming_folder)
if new_invoices isNoneorlen(new_invoices) == 0:
print("没有需要处理的发票")
return
# Step 2: 数据清洗和标准化
print("\n[Step 2/5] 🔧 数据标准化...")
# 统一日期格式
defnormalize_date(date_str):
if pd.isna(date_str):
returnNone
date_str = str(date_str).replace("年", "-").replace("月", "-").replace("日", "")
try:
return pd.to_datetime(date_str).strftime("%Y-%m-%d")
except:
return date_str
new_invoices["开票日期_标准"] = new_invoices["开票日期"].apply(normalize_date)
# 金额数值化
for col in ["金额(元)", "税额(元)", "价税合计(元)"]:
new_invoices[col] = pd.to_numeric(new_invoices[col], errors="coerce")
# Step 3: 自动分类归档
print("\n[Step 3/5] 📁 自动分类归档...")
archive_log = auto_classify_and_archive(
new_invoices, incoming_folder, archive_folder
)
# Step 4: 生成分析报表
print("\n[Step 4/5] 📊 生成分析报表...")
report_file = os.path.join(archive_folder,
f"发票分析报告_{datetime.now().strftime('%Y%m%d')}.xlsx")
with pd.ExcelWriter(report_file, engine="openpyxl") as writer:
# Sheet 1: 完整台账
new_invoices.to_excel(writer, sheet_name="发票台账", index=False)
# Sheet 2: 按销售方汇总
if"销售方"in new_invoices.columns:
by_seller = new_invoices[new_invoices["价税合计(元)"].notna()] \
.groupby("销售方")["价税合计(元)"].agg(["sum", "count"]) \
.round(2)
by_seller.columns = ["金额合计", "发票数量"]
by_seller = by_seller.sort_values("金额合计", ascending=False)
by_seller.to_excel(writer, sheet_name="按供应商汇总")
# Sheet 3: 按月份汇总
if"开票日期_标准"in new_invoices.columns:
new_invoices["月份"] = pd.to_datetime(
new_invoices["开票日期_standard"], errors="coerce"
).dt.to_period("M")
by_month = new_invoices[new_invoices["价税合计(元)"].notna()] \
.groupby("月份")["价税合计(元)"].agg(["sum", "count"])
by_month.columns = ["金额合计", "发票数量"]
by_month.to_excel(writer, sheet_name="按月度汇总")
# Sheet 4: 异常发票
abnormal = new_invoices[new_invoices["提取状态"] != "✅ 成功"]
iflen(abnormal) > 0:
abnormal.to_excel(writer, sheet_name="需人工复核", index=False)
# Step 5: 输出总结
elapsed = (datetime.now() - start_time).total_seconds()
print(f"\n[Step 5/5] ✅ 全部完成!")
print(f" ⏱️ 总耗时: {elapsed:.1f} 秒")
print(f" 📊 处理发票: {len(new_invoices)} 张")
print(f" 💰 总金额: {new_invoices['价税合计(元)'].sum():,.2f} 元")
print(f" 📁 报表文件: {report_file}")
print(f" 📁 归档目录: {archive_folder}")
# 使用示例
# full_invoice_workflow(
# incoming_folder="./新发票接收区/",
# archive_folder="./发票归档库/"
# )八、 避坑指南
pdfplumber 或指定编码 | ||
tabula-py 替代 | ||
pytesseract | ||
九、 总结
电子发票管理的核心原则：让机器做机器擅长的事（批量处理、精确匹配），让人做人擅长的事（异常审核、业务判断）。
下一篇文章是这个系列的收官之作——我们将把前面学的所有知识串联起来，搭建一个完整的财务自动化工作流。