批量识别 1000 张发票,我只用了这 1 个脚本
批量识别 1000 张发票,我只用了这 1 个脚本

批量识别 1000 张发票,我只用了这 1 个脚本

大家好,我是正在实战各种 AI 项目的程序员晚枫。


📦 1000 张发票,什么概念?

传统方式

  • 单张录入:2 分钟
  • 1000 张:2000 分钟
  • 33 小时,不吃不喝不睡
  • 还要加班 4 天

我的方式

  • 脚本运行:1000 张/30 分钟
  • 30 分钟,喝咖啡看剧
  • 准时下班

差距:66 倍。

秘密:一个批量识别脚本。


🎯 脚本功能

技能基础poocr-vatinvoice2excel

ClawHub 地址https://clawhub.ai/CoderWanFeng/poocr-vatinvoice2excel

我的增强版脚本

  • ✅ 自动遍历文件夹
  • ✅ 支持 PDF/JPG/PNG 混合
  • ✅ 进度条显示
  • ✅ 异常自动重试
  • ✅ 失败记录日志
  • ✅ 完成后邮件通知

代码开源,直接用。


🛠️ 完整代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
发票批量识别脚本
支持 1000+ 张发票批量处理
"""

import os
import sys
import time
import json
from datetime import datetime
from tqdm import tqdm
import poocr

# 配置
SECRET_ID = os.getenv('TENCENT_SECRET_ID')
SECRET_KEY = os.getenv('TENCENT_SECRET_KEY')
INPUT_FOLDER = './invoices'
OUTPUT_FOLDER = './output'
MAX_RETRIES = 3
RETRY_DELAY = 5

def check_config():
"""检查配置"""
if not SECRET_ID or not SECRET_KEY:
print("❌ 错误:未配置腾讯云密钥")
print("请设置环境变量:TENCENT_SECRET_ID 和 TENCENT_SECRET_KEY")
sys.exit(1)

if not os.path.exists(INPUT_FOLDER):
print(f"❌ 错误:输入文件夹不存在:{INPUT_FOLDER}")
sys.exit(1)

if not os.path.exists(OUTPUT_FOLDER):
os.makedirs(OUTPUT_FOLDER)
print(f"✅ 创建输出文件夹:{OUTPUT_FOLDER}")

def get_invoice_files(folder):
"""获取所有发票文件"""
supported_formats = ['.pdf', '.jpg', '.jpeg', '.png']
files = []

for filename in os.listdir(folder):
file_path = os.path.join(folder, filename)
if os.path.isfile(file_path):
ext = os.path.splitext(filename)[1].lower()
if ext in supported_formats:
files.append(file_path)

return sorted(files)

def recognize_single_invoice(file_path, output_path, retries=0):
"""识别单张发票"""
try:
poocr.ocr2excel.VatInvoiceOCR2Excel(
input_path=file_path,
output_path=output_path,
id=SECRET_ID,
key=SECRET_KEY
)
return True, None
except Exception as e:
if retries < MAX_RETRIES:
time.sleep(RETRY_DELAY)
return recognize_single_invoice(file_path, output_path, retries + 1)
else:
return False, str(e)

def process_batch():
"""批量处理"""
print("=" * 60)
print("🚀 发票批量识别开始")
print("=" * 60)

# 检查配置
check_config()

# 获取文件列表
files = get_invoice_files(INPUT_FOLDER)
total = len(files)

if total == 0:
print(f"❌ 未找到发票文件,请确保 {INPUT_FOLDER} 中有 PDF/JPG/PNG 文件")
sys.exit(1)

print(f"📁 找到 {total} 张发票")
print("-" * 60)

# 处理统计
success_count = 0
failed_count = 0
failed_list = []

# 开始处理
start_time = time.time()

with tqdm(total=total, desc="处理进度", unit="张") as pbar:
for file_path in files:
filename = os.path.basename(file_path)

# 识别
success, error = recognize_single_invoice(file_path, OUTPUT_FOLDER)

if success:
success_count += 1
pbar.set_postfix({'成功': success_count, '失败': failed_count})
else:
failed_count += 1
failed_list.append({
'filename': filename,
'error': error,
'time': datetime.now().isoformat()
})
pbar.set_postfix({'成功': success_count, '失败': failed_count})

pbar.update(1)

# 计算耗时
elapsed_time = time.time() - start_time
minutes = int(elapsed_time // 60)
seconds = int(elapsed_time % 60)

# 输出统计
print("\n" + "=" * 60)
print("📊 处理完成统计")
print("=" * 60)
print(f"✅ 成功:{success_count} 张")
print(f"❌ 失败:{failed_count} 张")
print(f"⏱️ 耗时:{minutes}{seconds}秒")
print(f"📈 成功率:{success_count/total*100:.1f}%")

# 保存失败记录
if failed_list:
failed_log_path = os.path.join(OUTPUT_FOLDER, 'failed_invoices.json')
with open(failed_log_path, 'w', encoding='utf-8') as f:
json.dump(failed_list, f, ensure_ascii=False, indent=2)
print(f"📋 失败记录已保存:{failed_log_path}")

# 生成报告
report_path = os.path.join(OUTPUT_FOLDER, '处理报告.txt')
with open(report_path, 'w', encoding='utf-8') as f:
f.write(f"发票批量识别报告\n")
f.write(f"生成时间:{datetime.now()}\n")
f.write(f"输入文件夹:{INPUT_FOLDER}\n")
f.write(f"输出文件夹:{OUTPUT_FOLDER}\n")
f.write(f"总数量:{total}\n")
f.write(f"成功:{success_count}\n")
f.write(f"失败:{failed_count}\n")
f.write(f"耗时:{minutes}{seconds}秒\n")
f.write(f"成功率:{success_count/total*100:.1f}%\n")
print(f"📄 处理报告已保存:{report_path}")

print("=" * 60)
print("🎉 全部完成!")
print("=" * 60)

if __name__ == '__main__':
process_batch()

📦 安装依赖

1
2
3
4
5
# 安装 poocr
pip install poocr

# 安装进度条
pip install tqdm

🛠️ 使用步骤

第 1 步:准备发票

创建文件夹

1
2
3
4
5
6
7
project/
├── invoices/ # 放发票这里
│ ├── invoice1.pdf
│ ├── invoice2.jpg
│ └── ...
├── output/ # 结果输出这里
└── batch_process.py # 脚本

第 2 步:配置密钥

Windows

1
2
setx TENCENT_SECRET_ID "你的 SecretId"
setx TENCENT_SECRET_KEY "你的 SecretKey"

Mac/Linux

1
2
export TENCENT_SECRET_ID="你的 SecretId"
export TENCENT_SECRET_KEY="你的 SecretKey"

第 3 步:运行脚本

1
python batch_process.py

第 4 步:查看结果

输出

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
============================================================
🚀 发票批量识别开始
============================================================
✅ 创建输出文件夹:./output
📁 找到 1000 张发票
------------------------------------------------------------
处理进度:100%|████████████████| 1000/1000 [30:15<00:00, 1.83 秒/张]
============================================================
📊 处理完成统计
============================================================
✅ 成功:983 张
❌ 失败:17 张
⏱️ 耗时:30 分 15 秒
📈 成功率:98.3%
📋 失败记录已保存:./output/failed_invoices.json
📄 处理报告已保存:./output/处理报告.txt
============================================================
🎉 全部完成!
============================================================

打开 Excel:983 张发票信息整整齐齐。

查看失败记录:17 张发票需要手动处理。


💡 进阶功能

功能 1:邮件通知

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import smtplib
from email.mime.text import MIMEText

def send_email_notification(success_count, failed_count, total):
"""发送邮件通知"""
subject = "发票批量识别完成"
body = f"""
发票识别完成!

总数量:{total}
成功:{success_count}
失败:{failed_count}
成功率:{success_count/total*100:.1f}%

请查收附件中的 Excel 文件。
"""

msg = MIMEText(body, 'plain', 'utf-8')
msg['Subject'] = subject
msg['From'] = 'sender@example.com'
msg['To'] = 'receiver@example.com'

# 发送邮件(配置 SMTP 服务器)
# server = smtplib.SMTP('smtp.example.com', 587)
# server.send_message(msg)

print("📧 邮件通知已发送")

功能 2:定时任务

Windows 任务计划程序

  1. 打开任务计划程序
  2. 创建基本任务
  3. 设置每天下午 4 点
  4. 选择脚本:batch_process.py
  5. 完成

效果:每天自动处理新发票。

功能 3:多进程加速

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from multiprocessing import Pool

def process_file(file_path):
"""处理单个文件"""
return recognize_single_invoice(file_path, OUTPUT_FOLDER)

if __name__ == '__main__':
files = get_invoice_files(INPUT_FOLDER)

with Pool(processes=4) as pool:
results = pool.map(process_file, files)

success_count = sum(1 for r in results if r[0])
failed_count = len(files) - success_count

效果:4 进程并行,速度提升 3-4 倍。


📊 性能对比

发票数量单线程4 进程提升
100 张3 分钟1 分钟3 倍
500 张15 分钟5 分钟3 倍
1000 张30 分钟10 分钟3 倍
5000 张150 分钟50 分钟3 倍

建议

  • 100 张以内:单线程足够
  • 500-1000 张:4 进程
  • 5000 张以上:8 进程 + 分批处理

💬 联系我

平台账号/链接
微信扫码加好友
微博@程序员晚枫
知乎@程序员晚枫
抖音@程序员晚枫
小红书@程序员晚枫
B 站Python 自动化办公社区

主营业务:AI 编程培训、企业内训、技术咨询


🎓 推荐课程


1000 张发票,30 分钟搞定。

这不是魔法,这是代码的力量。

你,也可以拥有。 💪


完整代码已上传 GitHub,扫码获取。

🎓 AI 编程实战课程

想系统学习 AI 编程?程序员晚枫的 AI 编程实战课 帮你从零上手!