# -*- coding: utf-8 -*- """ 普康健康_发票查验 """ # 加载模块 import hashlib import json import shutil import uuid from base64 import b64decode, b64encode from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path import cv2 import numpy import pandas from utils.client import Authenticator, HTTPClient, RequestException, restrict from utils.pandas_extension import open_csv, save_as_workbook, traverse_directory # 影像件压缩 def image_compression( image_path: str | None = None, image_format: str | None = None, image_data: bytes | None = None, # 数据类型为包含图像文件的二进制数据的字节串 image_size_specified: int = 2, # 指定影像件大小 raw: bool = False, # 250804新增返回是否为完整URI数据格式 ) -> str | None: try: # 若影像件路径数据类型为STR则创建路径对象 if isinstance(image_path, str): image_path = Path(image_path) # 影像件文件名称后缀 image_format = image_path.suffix.strip().lstrip(".").lower() # 读取影像件数据 with open(image_path, "rb") as image: image_data = image.read() # 影像件数据BASE64编码 image_data_base64 = b64encode(image_data).decode("utf-8") # 指定影像件大小的单位由MB转为KB image_size_specified = image_size_specified * 1024 * 1024 # 若影像件大小小于指定影像件大小则返回BASE64编码后影像件数据 if len(image_data_base64) < image_size_specified: if raw: # 返回非完整URI数据格式 return image_data_base64 else: # 返回完整URI数据格式 return f"data:image/{image_format};base64,{image_data_base64}" # OPENCV解码（数据类型为NUMPY-UINT8） image_data_cv2 = cv2.imdecode( numpy.frombuffer(image_data, numpy.uint8), cv2.IMREAD_COLOR ) # 若OPENCV解码失败则抛出异常 if image_data_cv2 is None: raise RuntimeError(f"OPENCV解码发生异常") # 初始化近似BASE64编码后影像件数据 proximate_image_data_base64 = None # 初始化最小压缩前后影像件大小差值 min_image_size_difference = float("inf") # 基于双层压缩方法：先外层降低图像质量，再内层缩小图像尺寸 for quality in range(90, 0, -10): image_data_cv2_ = image_data_cv2.copy() # 根据影像件格式匹配图片质量配置 # noinspection PyUnreachableCode match image_format: case "png": encoding_params = [cv2.IMWRITE_PNG_COMPRESSION, 10 - quality // 10] case _: encoding_params = [cv2.IMWRITE_JPEG_QUALITY, quality] for i in range(25): # 降低图像质量 # noinspection PyTypeChecker success, image_data_encoded = cv2.imencode( image_format, image_data_cv2_, encoding_params ) # 图像编码 # 若图像编码失败则退出 if not success: break image_data_base64 = b64encode(image_data_encoded.tobytes()).decode( "utf-8" ) # 压缩前后影像件大小差值 image_size_difference = len(image_data_base64) - image_size_specified if image_size_difference <= 0: if raw: return image_data_base64 else: return f"data:image/{image_format};base64,{image_data_base64}" if image_size_difference < min_image_size_difference: min_image_size_difference = image_size_difference proximate_image_data_base64 = image_data_base64 # 影像件高度和宽度 image_height, image_weight = image_data_cv2_.shape[:2] # 若仍超过影像件指定大小则调整图像尺寸 image_data_cv2_ = cv2.resize( image_data_cv2_, dsize=(int(image_weight * 0.9), int(image_height * 0.9)), interpolation=cv2.INTER_AREA, ) if proximate_image_data_base64: if raw: return proximate_image_data_base64 else: return f"data:image/{image_format};base64,{image_data_base64}" else: raise RuntimeError("影像件压缩失败") except: return None # 票据查验接口（需要） @restrict(refill_rate=5, max_tokens=5) # 限速至5QPS def invoices_verification( image_index, image_path=None, invoice_number=None, invoice_code=None, invoice_check_code=None, invoice_date=None, invoice_amount=None, id_number=None, process_mode=None, supplier=None, ): try: # 若影像件地址非空则imgBASE64请求，否则根据发票五要素请求 if image_path: match process_mode: case "通过影像件本地地址": # 创建路径对象 image_path = Path(image_path) # 影像件文件名称后缀 image_format = image_path.suffix.strip().lstrip(".").lower() # 读取影像件数据 with open(image_path, "rb") as image: image_data = image.read() case "通过影像件对象服务器地址": image_format, image_data = http_client.download(url=image_path) # 断定影像件格式为JGP、JPEG或者PNG # noinspection PyUnboundLocalVariable assert image_format in [ "jpg", "jpeg", "png", ], f"影像件格式({image_format})不支持" match supplier: case "szkt": image_data_base64 = image_compression( image_format=image_format, image_data=image_data ) # noinspection PyUnusedLocal response = http_client.post( # 深圳快瞳增值税发票、医疗发票查验兼容版 url="https://ai.inspirvision.cn/s/api/ocr/invoiceCheckAll", # 用于和深圳快瞳联查时定位请求 headers={"X-RequestId-Header": image_index}, data={ "token": authenticator.get_token(servicer="szkt"), "imgBase64": image_data_base64, }, ) case "bjfd": image_data_base64 = image_compression( image_format=image_format, image_data=image_data, raw=True ) # 北京分单不支持完整URI数据格式 # 业务入参，序列化并BASE64编码 data = b64encode( json.dumps( { "fileByte": image_data_base64, "fileType": ( "png" if image_format == "png" else "jpg" ), # 北京分单影像件格式支持JPG、PNG或PDF（本脚本暂不支持PDF） } ).encode("utf-8") ).decode("utf-8") # 应用账号 appid = "mbYr11Rc_42" # 随机标识 noise = image_index # 版本号 version = "1.0" # 装配签名 sign = ( hashlib.md5( f"appid={appid}&data={data}&noise={noise}&key=80357535c95333c3b133dfe5533f6334fe5e9321&version={version}".encode( "utf-8" ) ) .hexdigest() .upper() ) # noinspection PyUnusedLocal response = http_client.post( # 北京分单增值税发票、医疗票据二维码查验接口 url="https://api.fendanyun.com/rsx/api/checkByQRCode", headers={"Content-Type": "application/json; charset=utf-8"}, json={ "appid": appid, "data": data, "noise": noise, "version": version, "sign": sign, }, ) else: response = http_client.post( # 深圳快瞳增值税发票、医疗发票查验兼容版 url="https://ai.inspirvision.cn/s/api/ocr/invoiceCheckAll", data={ "token": authenticator.get_token(servicer="szkt"), "invoiceNumber": invoice_number, "invoiceCode": invoice_code, "checkCode": invoice_check_code, "invoicingDate": invoice_date, "pretaxAmount": invoice_amount, "idCardNo": id_number, }, ) except RequestException as request_exception: response = { "status": request_exception.status, "code": request_exception.code, "message": request_exception.message, } except Exception as exception: response = { "code": "40000", "message": f"发生其它异常{exception}", } return image_index, response if __name__ == "__main__": print("已启动批量票据查验") match input("请选择票据查验供应商(1:深圳快瞳，2:北京分单，其它任意字符:退出脚本):"): case "1": supplier = "szkt" case "2": supplier = "bjfd" case _: print("选择退出脚本!") exit(0) match input( "请选择处理流程(1:批量解析已归档响应报文，2:根据影像件地址或票据五要素批量查验，其它任意字符:退出脚本):" ): case "1": # 打开前置影像件索引CSV文件 dataframe = open_csv(file_name="dataframe_indexed.csv") case "2": print("正在归档响应报文...", end="") # 创建响应报文目录路径对象 responses_path = Path("temporary/responses") # 若响应报文目录路径不存在则创建 if not responses_path.exists(): responses_path.mkdir(parents=True, exist_ok=True) # 创建归档响应报文目录路径对象 archives_path = Path("temporary/archives") # 若归档响应报文目录路径不存在则创建 if not archives_path.exists(): archives_path.mkdir(parents=True, exist_ok=True) # 遍历响应报文目录下所有文件名后缀为JSON的文件路径 for file_path in Path(responses_path).glob("*.json"): # 若文件路径为文件 if file_path.is_file(): # 移动响应报文由响应报文目录至归档响应报文目录 shutil.move(str(file_path), str(archives_path / file_path.name)) print("已完成") match input( "请选择批量查验方法(1:通过影像件本地地址，2:通过影像件对象服务器地址，3:通过增值税发票和医疗票据的五要素，其它任意字符:退出脚本):" ): case "1": print("正在读取影像件本地地址...", end="") dataframe = traverse_directory( directory_path="待查验发票", suffixes=[".jpg", ".jpeg", ".png"] ) # 修改列名相对路径为影像件地址 dataframe.rename(columns={"相对路径": "影像件地址"}, inplace=True) process_mode = "通过影像件本地地址" case "2": print("正在读取影像件对象服务器地址...", end="") dataframe = open_csv(file_name="dataframe.csv") # 断定列名包括赔案编号、发票编号和影像件地址 assert all( [ column_name in dataframe.columns for column_name in ["赔案编号", "发票编号", "影像件地址"] ] ), "CSV文件中列名必须包括赔案编号、发票编号和影像件地址" # 根据赔案编号和发票编号去重 dataframe.drop_duplicates( subset=["赔案编号", "发票编号"], keep="first", inplace=True ) # 处理方式 process_mode = "通过影像件对象服务器地址" case "3": print("正在读取增值税发票和医疗票据的五要素...", end="") dataframe = open_csv(file_name="dataframe.csv") # 断定列名包括身份证号码后六位、发票编号、发票代码、校验号码后六位、开票日期和发票金额 assert all( [ column_name in dataframe.columns for column_name in [ "身份证号码后六位", "发票编号", "发票代码", "校验号码后六位", "开票日期", "发票金额", ] ] ), "CSV文件中列名必须包括身份证号码后六位、发票编号、发票代码、校验号码后六位、开票日期和发票金额" # 根据身份证号码后六位、发票编号、发票代码、校验号码后六位、开票日期和发票金额去重 dataframe.drop_duplicates( subset=[ "身份证号码后六位", "发票编号", "发票代码", "校验号码后六位", "开票日期", "发票金额", ], keep="first", inplace=True, ) # 格式化开票日期 dataframe["开票日期"] = dataframe["开票日期"].str.replace( "-", "", regex=False ) # 处理方式 process_mode = "通过增值税发票和医疗票据的五要素" case _: print("选择退出脚本!") exit(0) # 统计待查验发票张数 rows = dataframe.shape[0] # 若待查验发票张数为0则退出脚本 if rows == 0: print("待查验发票张数为0，退出脚本") exit(0) print(f"已完成，待查验发票张数为 {rows}") # 添加索引 dataframe["索引"] = dataframe.apply( lambda x: uuid.uuid4().hex, axis="columns" ) dataframe.to_csv("dataframe_indexed.csv", index=False) # 创建深圳快瞳获取访问令牌方法 authenticator = Authenticator() # 初始化请求客户端 http_client = HTTPClient() # 用于记录已完成任务数 completed_futures = 0 # 创建线程池 with ThreadPoolExecutor(max_workers=5) as executor: # noinspection PyUnreachableCode # noinspection PyUnboundLocalVariable match process_mode: case "通过影像件本地地址" | "通过影像件对象服务器地址": futures = [ executor.submit( invoices_verification, image_index=row.索引, image_path=row.影像件地址, process_mode=process_mode, supplier=supplier, ) for row in dataframe[["索引", "影像件地址"]].itertuples( index=False, name="row" ) ] case "通过增值税发票和医疗票据的五要素": # 提交任务 futures = [ executor.submit( invoices_verification, image_index=row.索引, invoice_number=row.发票编号, invoice_code=row.发票代码, invoice_check_code=row.校验号码后六位, invoice_date=row.开票日期, invoice_amount=row.发票金额, id_number=row.身份证号码后六位, process_mode=process_mode, supplier=supplier, ) for row in dataframe[ [ "索引", "发票编号", "发票代码", "校验号码后六位", "开票日期", "发票金额", "身份证号码后六位", ] ].itertuples(index=False, name="row") ] for future in as_completed(futures): index, response = future.result() # 保存报文 with open( "temporary/responses/{}.json".format(index), "w", encoding="utf-8", ) as file: json.dump(response, file, ensure_ascii=False) completed_futures += 1 print(f"已完成 {completed_futures / rows * 100:.2f} %") case _: print("选择退出脚本!") exit(0) print("正在解析报文...", end="") # 解析后数据体 dataframe_parsed = [] # 遍历报文所在目录 for path_object in list(Path("temporary/responses").glob("*.json")): # 解析报文结构 parse = { "索引": "", "机打发票号码": "", "发票金额": "", "购买方": "", "销售方": "", "发票状态": "", "最大销售项目名称": "", "最大销售项目数量": "", "XML版式文件": "", } # 若路径对象包含下划线则在解析报文结构添加赔案编号和发票编号 if "_" in path_object.stem: parse["赔案编号"] = path_object.stem.split("_")[0] parse["发票编号"] = path_object.stem.split("_")[1] # 打开报文并JSON逆序列化 with open(path_object, "r", encoding="utf-8") as file: response = json.load(file) # 索引 parse["索引"] = path_object.stem match supplier: case "szkt": try: # 响应状态码 status_code = response.get("status", "") # 错误码 code = response.get("code", "") # 流水号 serial = response.get("serialNo", "") # 若响应状态码为200且错误码为10000，则定义为响应成功 if status_code == 200 and code == 10000: # 查验类型，若查验类型为003081则为医疗票据查验，003082则为增值税发票查验，两者报文结构不一致 match response.get("data").get( "productCode" ): # 若响应成功则必定存在键DATA和PRODUCTCODE # 解析医疗票据查验结果 case "003081": parse["机打发票号码"] = response.get("data").get( "billNumber" ) parse["校验码"] = response.get("data").get("checkCode") parse["发票金额"] = response.get("data").get("amount") parse["购买方"] = response.get("data").get("payer") parse["销售方"] = response.get("data").get("payeeName") # 发票状态 match response.get("data").get("flushedRed"): case "true": parse["发票状态"] = "正常" case "false": parse["发票状态"] = "已红冲" # 最大销售项目数量 max_item_quantity = 0 # 遍历销售项目列表 for item in response.get("data").get("feeitems", []): # 销售项目数量 item_quantity = item.get("number") # 若销售项目数量非空，进一步判断是否包含斜杠，若包含斜杠则分割并取第一部分，最后转为浮点 if item_quantity: if "/" in item_quantity: item_quantity = item_quantity.split("/")[0] item_quantity = float(item_quantity) else: item_quantity = 1 if item_quantity > max_item_quantity: parse["最大销售项目名称"] = item.get( "itemName", "" ) parse["最大销售项目数量"] = str(item_quantity) parse["XML版式文件"] = response.get("PDFInfo", {}).get( "fileUrl" ) # 解析增值税发票查验结果 case "003082": parse["机打发票号码"] = ( response.get("data").get("details").get("number") ) parse["校验码"] = ( response.get("data") .get("details") .get("check_code") ) parse["发票金额"] = ( response.get("data").get("details").get("total") ) parse["购买方"] = ( response.get("data").get("details").get("buyer") ) parse["销售方"] = ( response.get("data").get("details").get("seller") ) # 发票状态 match response.get("data").get("details").get( "invoiceTypeNo" ): case "0": parse["发票状态"] = "正常" case "1": parse["发票状态"] = "无法查验" case "2" | "3" | "7" | "8": parse["发票状态"] = "已红冲" max_item_quantity = 0 items = ( response.get("data").get("details").get("items", []) ) for item in items: item_quantity = ( float(item.get("quantity", 1)) if item.get("quantity") else 1 ) if item_quantity > max_item_quantity: parse["最大销售项目名称"] = item.get("name") parse["最大销售项目数量"] = str(item_quantity) # XML版式文件（25-06-11本接口不在提供版式文件，通过另一接口可获取数电增值税发票版式文件） parse["XML版式文件"] = "本接口不再提供版式文件" # 若响应状态码为400且错误码为10001或10100，则定义为假票 elif status_code == 400 and (code == 10001 or code == 10100): parse["发票状态"] = "假票" else: raise Exception("解析报文发生其它异常") except Exception as exception: parse["发票状态"] = "{}".format(response.get("message")) case "bjfd": try: # 不验签，业务出参BASE64解码并反序列化 response = json.loads( b64decode(response.get("data")).decode("utf-8") ) # 增值税发票、医疗票据查验结果BASE64解码并反序列化 response["message"] = json.loads( b64decode(response.get("message")).decode("utf-8") ) # 错误码 code = response.get("result") # 流水号 serial = response.get("message").get("checkId") # 核验结果代码 result_code = response.get("message").get("resultCode") # 若错误码为S0000则定义为响应成功 if code == "S0000": # noinspection PyUnreachableCode match result_code: # 若查验成功则根据增值税发票、医疗票据状态匹配发票状态 case "200": parse["机打发票号码"] = ( response.get("message") .get("tickMainInfo") .get("invoiceNo") ) parse["发票金额"] = ( response.get("message") .get("tickMainInfo") .get("invoiceTotalPrice") ) parse["购买方"] = ( response.get("message") .get("tickMainInfo") .get("payerPartyName") ) parse["销售方"] = ( response.get("message") .get("tickMainInfo") .get("invoicingPartyName") ) max_item_quantity = 0 for item in ( response.get("message") .get("tickMainInfo") .get("chargeItems", []) ): item_quantity = ( float(item.get("num", 1)) if item.get("num") else 1 ) if item_quantity > max_item_quantity: parse["最大销售项目名称"] = item.get( "chargeName" ) parse["最大销售项目数量"] = str(item_quantity) match response.get("message").get("invoiceStatus"): case "0": parse["发票状态"] = "正常" case ( "1" | "2" ): # 沿用深圳快瞳解析规则，北京分单已开红票和已作废映射为已红冲 parse["发票状态"] = "已红冲" case "E20003" | "E20007 ": parse["发票状态"] = "假票" # 其它情况发票状态根据核验结果描述 case _: parse["发票状态"] = response.get("message").get( "resultMsg" ) except Exception as exception: parse["发票状态"] = str(exception) dataframe_parsed.append(parse) dataframe_parsed = pandas.DataFrame(data=dataframe_parsed, dtype=str) # 将解析数据集拼接至数据集 dataframe = dataframe.merge(right=dataframe_parsed, how="left", on=["索引"]) # 填补缺失值 dataframe = dataframe.fillna(value="") print("已完成") print("正在保存为工作簿...", end="") save_as_workbook(worksheets=[("Sheet1", dataframe)], workbook_name="results.xlsx") print("已完成")