Python制作发票自动校核微信机器人

快来打我* 2023-07-15 13:16 13阅读 0赞

**制作初衷：**

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

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

**源码地址：**[https://github.com/yc2code/WechatInvoiceParser][https_github.com_yc2code_WechatInvoiceParser]

**P.S. 工具基于微信网页版，因为微信官方对于账号有限制，新建的账号可能无法使用，会报：`KeyError: 'pass_ticket'`，如图：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pYm9ubm4_size_16_color_FFFFFF_t_70]**所以工具只能使用注册时间较早的账号**

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

**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"  # 默认购方税号
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pYm9ubm4_size_16_color_FFFFFF_t_70 1]  
**另外，代码在运行时会在同文件夹下创建一个Picture的文件夹，用于存储待解析的图片，会创建 work\_log.csv 文件，用于存储识别信息的记录，还有 system\_log.text 用于输出运行相应的日志。**

由于本身需求较少，所以以上代码功能相对单薄，仅仅作为一个辅助的小脚本使用。若要进行优化完善，wxpy库提供了很多丰富的功能，可以在此基础上打造更加合理完善的，符合个性化需求的微信机器人。

水平有限，难免存在错误，还望指正。

[https_github.com_yc2code_WechatInvoiceParser]: https://github.com/yc2code/WechatInvoiceParser.git
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pYm9ubm4_size_16_color_FFFFFF_t_70]: /images/20230528/2bd8e930ceef4a26b8299aec172b6f12.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pYm9ubm4_size_16_color_FFFFFF_t_70 1]: /images/20230528/f8e4718c53d34bd6841681939e40bce2.png