制作初衷:
本代码使用简单的封装方法,并做了比较走心的注释,希望能给初学Python的小伙伴提供一些灵感,也能让有实际需求的人可以快速修改、使用。
源码地址:https://github.com/yc2code/WechatInvoiceParser
P.S. 工具基于微信网页版,因为微信官方对于账号有限制,新建的账号可能无法使用,会报:KeyError: 'pass_ticket'
,如图:
所以工具只能使用注册时间较早的账号
发票自动校核微信机器人代码部分
1. 工具文件 – Utils
包含三个部分:发票校核类 Invoice、解析数据类 DataParser 和推送日志类 Pushover
# -*- coding: utf-8 -*-
# Utils.py
import base64
import csv
import os
import time
import requests
from Config import config
class Invoice:
"""
发票识别类
使用百度发票识别API,免费使用
官方地址 https://ai.baidu.com/docs#/OCR-API/5099e085
其它功能及配置请移步官网
"""
@staticmethod
def get_pic_content(image_path):
"""
方法--打开图片
以二进制格式打开
"""
with open(image_path, 'rb') as pic:
return pic.read()
@staticmethod
def parse_invoice(image_binary):
"""
方法--识别图片
调用百度接口,返回识别后的发票数据
以下内容基本根据API调用的要求所写,无需纠结
各类报错码在官网文档可查
百度API注册及使用教程:http://ai.baidu.com/forum/topic/show/867951
"""
# 识别质量可选high及normal
# normal(默认配置)对应普通精度模型,识别速度较快,在四要素的准确率上和high模型保持一致,
# high对应高精度识别模型,相应的时延会增加,因为超时导致失败的情况也会增加(错误码282000)
access_token = "你的access_token"
api_url = f"https://aip.baidubce.com/rest/2.0/ocr/v1/vat_invoice?access_token={access_token}"
quality = "high"
header = {"Content-Type": "application/x-www-form-urlencoded"}
# 图像数据,base64编码后进行urlencode,要求base64编码和urlencode后大小不超过4M,
# 最短边至少15px,最长边最大4096px,支持jpg/jpeg/png/bmp格式
image_data = base64.b64encode(image_binary)
try:
data = {"accuracy": quality, "image": image_data}
response = requests.post(api_url, data=data, headers=header)
if response.status_code != 200:
print(time.ctime()[:-5], "Failed to get info")
return None
else:
result = response.json()["words_result"]
invoice_data = {
'检索日期': '-'.join(time.ctime().split()[1:3]),
'发票代码': result['InvoiceCode'],
'发票号码': result['InvoiceNum'],
'开票日期': result['InvoiceDate'],
'合计金额': result['TotalAmount'],
'价税合计': result['AmountInFiguers'],
'销售方名称': result['SellerName'],
'销售方税号': result['SellerRegisterNum'],
'购方名称': result['PurchaserName'],
'购方税号': result['PurchaserRegisterNum'],
"发票类型": result["InvoiceType"]
}
return invoice_data
except:
message = "发票识别API调用出现错误"
Pushover.push_message(message)
return None
finally:
print(time.ctime()[:-5], "产生一次了调用")
@staticmethod
def save_to_csv(invoice_data):
"""
方法--日志保存
将识别记录写入文件夹下work_log.csv文件
若无此文件则自动创建并写入表头
"""
if "work_log.csv" not in os.listdir():
not_found = True
else:
not_found = False
with open('./work_log.csv', 'a+') as file:
writer = csv.writer(file)
if not_found:
writer.writerow(invoice_data.keys())
writer.writerow(invoice_data.values())
@staticmethod
def run(image_path):
"""
主方法
解析完成返回信息,否则返回None
"""
image_binary = Invoice.get_pic_content(image_path)
invoice_data = Invoice.parse_invoice(image_binary)
if invoice_data:
Invoice.save_to_csv(invoice_data)
return invoice_data
return None
class DataParser:
"""
数据分析类
对识别返回后的数据进行整理,并于默认信息对比,查看有无错误
这里只简单实现整理信息和检查名称和税号的方法,有兴趣可以增加其他丰富的方法
"""
def __init__(self, invoice_data):
self.invoice_data = invoice_data
def get_detail_message(self):
"""
对得到的发票信息的格式进行整理
:return: 返回整理好的发票信息
"""
values = [value for value in self.invoice_data.values()]
detail_mess = f"完整信息为:" \
f"\n发票代码: {values[1]}\n发票号码: {values[2]}\n开票日期: {values[3]}" \
f"\n合计金额: {values[4]}\n价税合计: {values[5]}\n销售方名称: {values[6]}" \
f"\n销售方税号: {values[7]}\n购方名称: {values[8]}\n购方税号:{values[9]}"
return detail_mess
def get_brief_message(self):
"""
将信息中的名称和税号和默认值进行对比
只做对错判断,读者丰富一下可以增加指出错误位置的信息
:return: 返回判断的信息
"""
if self.invoice_data["购方名称"] == config["company_name"]:
brief_mess = "购方名称正确"
else:
brief_mess = "!购方名称错误!"
if self.invoice_data["购方税号"] == config["company_tax_number"]:
brief_mess += "\n购方税号正确"
else:
brief_mess += "\n!购方税号错误!"
return brief_mess
def parse(self):
brief_mess = self.get_brief_message()
detail_mess = self.get_detail_message()
return brief_mess, detail_mess
class Pushover:
"""
消息推送类
本次使用Pushover为推送消息软件(30 RMB,永久,推荐)
官网 https://pushover.net/
可以向微信一样把相关信息推送至不同设备
如果不需要可以把相关代码注释掉
"""
@staticmethod
def push_message(message):
message += ">>>来自Python发票校验"
try:
requests.post("https://api.pushover.net/1/messages.json", data={
"token": "你的Token",
"user": "你的User",
"message": message
})
except Exception as e:
print(time.ctime()[:-5], "Pushover failed", e, sep="\n>>>>>>>>>>\n")
2. 微信机器人文件 – Wechat
包含一个部分:微信处理类 Wechat
作用是初始化机器人,对微信的消息进行处理,分析并作出回应。
# -*- coding: utf-8 -*-
# Wechat.py
import os
from wxpy import *
class Wechat:
"""
微信处理类
对微信的消息进行处理,分析并作出回应
"""
def __init__(self, group_name, admin_name):
self.bot = Bot() # 类被实例化的时候即对机器人实例化
self.group_name = group_name # 指定群聊名
self.admin_name = admin_name # 管理员微信名
self.received_mess_list = [] # 过滤后的消息列表
self.order_list = [] # 管理命令列表
self.pic_list = [] # 待解析图片绝对路径列表
def get_group_mess(self):
"""
方法--获取消息
获取所有正常消息,进行过滤后存进消息列表
"""
# 调用此方法时先清空上次调用时列表所存储的数据
self.received_mess_list = []
for message in self.bot.messages:
# 如果为指定群聊或管理员的消息,存入group_mess
sender = message.sender.name
# >>>这里有一点要注意,如果你是用一个微信作为机器人且作为管理员<<<
# >>>然后用这个微信号在群聊发消息,则信息sender会之指向自己而不是群聊<<<
# >>>建议使用单独一个微信号作为机器人
if sender == self.group_name or sender == self.admin_name:
self.received_mess_list.append(message)
# 其他的消息过滤掉
self.bot.messages.remove(message)
return None
def parse_mess(self):
"""
方法--处理群聊消息
过滤获得的指定群聊消息
设定所有新增群聊图片的绝对路径及群聊中产生的文字命令
"""
# 调用此方法时先清空上次调用时列表所存储的数据
self.pic_list = []
self.order_list = []
# self.group_order = []
for message in self.received_mess_list:
# 如果信息类型为图片,则保存图片并添加到图片列表
if message.type == 'Picture' and message.file_name.split('.')[-1] != 'gif':
self.pic_list.append(Wechat.save_file(message))
# 如果消息类型为文字,则视为命令,保存到命令列表中
if message.type == 'Text':
self.order_list.append(message)
return None
@staticmethod
def save_file(image):
"""
方法--存储图片
这里使用静态方法,是因为本方法和类没有内部交互,静态方法可以方便其他程序的调用
解析名称,设定绝对路径,存储
:param image: 接收到的图片(可以看成是wxpy产生的图片类,它具有方法和属性)
:return: 返回图片的绝对路径
"""
path = os.getcwd()
# 如果路径下没有Pictures文件夹,则创建,以存放接收到的待识别图片
if "Pictures" not in os.listdir():
os.mkdir("Pictures")
# 设定一个默认的图片格式后缀
file_postfix = "png"
try:
# 尝试把图片的名称拆分,分别获取名称和后缀
file_name, file_postfix = image.file_name.split('.')
except Exception:
# 当然有时候可能拆分不了,就把默认的后缀给它
file_name = image.file_name
# 赋予绝对路径
file_path = path + '/Pictures/' + file_name + '.' + file_postfix
# 将图片存储到指定路径下
image.get_file(file_path)
return file_path
def send_group_mess(self, message):
"""
方法--发送群消息
:param message: 需要发送的内容
"""
try:
# 如果群聊名称被改变,搜索时会报错,如果找不到群聊,消息不会发送
group = self.bot.groups().search(self.group_name)[0]
group.send(message)
except IndexError:
print("找不到指定群聊,信息发送失败")
return None
def send_parse_log(self):
"""
方法--发送查询日志
向群聊内发送查询日志
"""
try:
# 如果群聊名称被改变,搜索时会报错,如果找不到群聊,消息不会发送
group = self.bot.groups().search(self.group_name)[0]
except IndexError:
print("找不到指定群聊,查询日志发送失败")
return None
try:
group.send_file("./work_log.csv")
except:
group.send("Oops, no log yet")
return None
def send_system_log(self):
"""
方法--发送系统日志
向群聊内发送查询日志
"""
try:
# 如果群聊名称被改变,搜索时会报错,如果找不到群聊,消息不会发送
group = self.bot.groups().search(self.group_name)[0]
except IndexError:
print("找不到指定群聊,系统日志发送失败")
return None
try:
group.send_file("./system_log.text")
except:
group.send("System log not found")
return None
3. 主文件 – Main
包含一个main函数,一部分为发票识别和处理,另一部分对于指令做出反应。
# -*- coding: utf-8 -*-
# Main.py
import time
from Utils import Invoice, DataParser
from Config import config
from Wechat import *
# Author : 达希
# Email : way2go.dash@gmail.com
def main():
"""
主方法
一部分为发票识别和处理,另一部分对于指令做出反应
"""
# 输出重定向,将print语句都写进系统日志文件
file = open("./system_log.text", "a+")
sys.stdout = file
# 实例化微信机器人,传入群聊名和管理员名
wechat = Wechat(config["group_name"], config["admin_name"])
while True:
time.sleep(1)
wechat.get_group_mess()
wechat.parse_mess()
# 若群聊有要处理的图片,则迭代解析
if wechat.pic_list:
for pic in wechat.pic_list:
invoice_data = Invoice.run(pic)
if invoice_data:
data_parser = DataParser(invoice_data)
brief_mess, detail_mess = data_parser.parse()
wechat.send_group_mess(detail_mess) # 先发送发票识别详细信息
time.sleep(0.5)
wechat.send_group_mess(brief_mess) # 返回名称和税号是否有错误
else:
wechat.send_group_mess("请求未成功,请重试或联系管理员")
# 若有相关命令,则做出相应反应
if wechat.order_list:
for order in wechat.order_list:
if "开票信息" in order.text:
wechat.send_group_mess(config["company_name"])
time.sleep(0.5)
wechat.send_group_mess(config["company_tax_number"])
elif "SEND LOG" in order.text:
wechat.send_parse_log()
elif "SEND SYSTEM LOG" in order.text:
wechat.send_system_log()
elif "BREAK" in order.text:
wechat.send_group_mess("收到关机指令,正在关机")
file.close()
return None
if __name__ == "__main__":
main()
4. 配置文件 – Config
包含微信的配置文件信息
config = {
"group_name": "发票校核ASAP", # 校核群聊名称,由于本代码默认没有同名群聊,所以建议设为复杂值
"admin_name": "达希", # 管理员微信名(非备注)
"company_name": "代码网络技术无限公司", # 默认购方名称
"company_tax_number": "XXX00000000000XXX" # 默认购方税号
}
另外,代码在运行时会在同文件夹下创建一个Picture的文件夹,用于存储待解析的图片,会创建 work_log.csv 文件,用于存储识别信息的记录,还有 system_log.text 用于输出运行相应的日志。
由于本身需求较少,所以以上代码功能相对单薄,仅仅作为一个辅助的小脚本使用。若要进行优化完善,wxpy库提供了很多丰富的功能,可以在此基础上打造更加合理完善的,符合个性化需求的微信机器人。
总结
到此这篇关于Python制作发票自动校核微信机器人的文章就介绍到这了,更多相关Python制作发票自动校核微信机器人内容请搜索python博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持python博客!
标签:requests
Powered By python教程网 鲁ICP备18013710号
python博客 - 小白学python最友好的网站!