Python实现发票自动校核微信机器人的方法

制作初衷：

外地开了票到公司后发现信息有错误，无法报销；
公司的行政和财务经常在工作日被问及公司开票信息，影响心情和工作；
引入相应的专业APP来解决发票问题对于一般公司成本较高；
看到朋友孟要早睡写过脚本来解决这个问题，但因为公司场景不相同，无法复用，所以新写了一个

本代码使用简单的封装方法，并做了比较走心的注释，希望能给初学Python的小伙伴提供一些灵感，也能让有实际需求的人可以快速修改、使用。

源码地址：https://github.com/yc2code/WechatInvoiceParser

P.S. 工具基于微信网页版，因为微信官方对于账号有限制，新建的账号可能无法使用，会报：KeyError: \'pass_ticket\'，如图：

所以工具只能使用注册时间较早的账号

发票自动校核微信机器人代码部分

1. 工具文件 ? Utils
包含三个部分：发票校核类 Invoice、解析数据类 DataParser 和推送日志类 Pushover

Invoice 调用的百度API，上传图片信息，得到解析数据；
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\" # 默认购方税号
}