jayus/VerifVATInvoice
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(invoice): 增加发票验证功能

Browse Source 
- 新增发票验证相关接口和页面
- 实现发票复验功能
- 添加发票列表和验证日志查询接口
- 更新发票状态和描述信息
- 优化文件上传逻辑，支持文件重复检测
This commit is contained in:
liuxiaoqing 2025-08-18 17:45:36 +08:00
parent ceb0b33656
commit 813a317ff6
6 changed files with 255 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

192
alibabacloud_sample/dataservice.py
View File

@ -28,19 +28,26 @@ class DataService:
return rows[0][0]
return file_path
# 根据文件路径获取hash
def get_invoice(self, file_path: str) -> dict[Any, Any]:
def get_invoice(self, file_path: str = None,invice_id: int = None) -> dict[Any, Any]:
conn = DBService().get_conn()
cursor = conn.cursor()
cursor.execute("""
select checkCode, drawer, formType, invoiceAmountPreTax, invoiceCode,
sql = """
select id, checkCode, drawer, formType, invoiceAmountPreTax, invoiceCode,
invoiceDate, invoiceNumber, invoiceTax, invoiceType, machineCode,
passwordArea, printedInvoiceCode, printedInvoiceNumber,
purchaserBankAccountInfo, purchaserContactInfo, purchaserName,
purchaserTaxNumber, recipient, remarks, reviewer,
sellerBankAccountInfo, sellerContactInfo, sellerName, sellerTaxNumber,
specialTag, title, totalAmount, totalAmountInWords,file_path from invoice where file_path = ?
specialTag, title, totalAmount, totalAmountInWords,file_path,verify_time,cyjgxx,
verify_time,verify_status,desc,desc_files,status
from invoice
"""
, (file_path,))
if invice_id:
sql += " where id = ?"
cursor.execute(sql, (invice_id,))
else:
sql += " where file_path = ?"
cursor.execute(sql, (file_path,))
rows = cursor.fetchall()
data = {}
if len(rows) > 0:
@ -48,6 +55,23 @@ class DataService:
data[cursor.description[i][0]] = rows[0][i]
return data
# 根据文件hash获取文件
def get_file(self, file_hash: str,file_size: int) -> dict[Any, Any] | None:
conn = DBService().get_conn()
cursor = conn.cursor()
cursor.execute("""
select file_name,file_path,file_size,file_type,file_time,file_hash
from file_data where file_hash = ? and file_size = ?
"""
, (file_hash,file_size,))
rows = cursor.fetchall()
if len(rows) > 0:
data = {}
for i in range(len(rows[0])):
data[cursor.description[i][0]] = rows[0][i]
return data
return None
# 插入发票记录
def insert_invoice_log(self, data: dict, file_path: str):
conn = DBService().get_conn()
@ -99,7 +123,52 @@ class DataService:
conn.commit()
invoice_id = cursor.lastrowid
return invoice_id
# 更新发票状态
def update_invoice_status(self, invoice_id: int, verify_status: str):
conn = DBService().get_conn()
cursor = conn.cursor()
cursor.execute(
"UPDATE invoice SET verify_time = ?,"
"verify_status = ?"
"WHERE id = ?",
(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), verify_status, invoice_id))
conn.commit()
def update_invoice_desc(self, invoice_id: int|str, desc: str, desc_file: str,status: str):
conn = DBService().get_conn()
cursor = conn.cursor()
cursor.execute(
"UPDATE invoice SET "
"desc = ?,"
"desc_files = ?,"
"status = ? "
"WHERE id = ?",
(desc, desc_file,status, invoice_id))
conn.commit()
# 查询验证记录
def get_verify_log(self, file_path: str) -> dict[Any, Any]:
conn = DBService().get_conn()
cursor = conn.cursor()
cursor.execute("""
select inspectionAmount,cyjgxx,verify_time,file_path from verify_log where file_path = ? order verify_time desc limit 1
"""
, (file_path,))
rows = cursor.fetchall()
data = {}
if len(rows) > 0:
for i in range(len(rows[0])):
data[cursor.description[i][0]] = rows[0][i]
return data
# 插入验证记录
def insert_verify_log(self, inspection_amount: str, cyjgxx: str, file_path: str):
conn = DBService().get_conn()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO verify_log(inspectionAmount,cyjgxx,verify_time,file_path) values(?,?,?,?)",
(inspection_amount,cyjgxx,parser.parse(datetime.now().strftime("%Y-%m-%d %H:%M:%S")),file_path))
conn.commit()
# 查询验证记录
def get_verify_log_list(self, file_path: str) -> list[dict[Any, Any]]:
conn = DBService().get_conn()
cursor = conn.cursor()
cursor.execute("""
@ -107,16 +176,109 @@ class DataService:
"""
, (file_path,))
rows = cursor.fetchall()
data = {}
if len(rows) > 0:
for i in range(len(rows[0])):
data[cursor.description[i][0]] = rows[0][i]
data = []
for row in rows:
data.append({
'inspectionAmount': row[0],
'cyjgxx': row[1],
'verify_time': row[2],
'file_path': row[3]
})
return data
def insert_verify_log(self, inspection_amount: str, cyjgxx: str, file_path: str):
# 查询发票列表
def get_invoice_list(self, value: str, verify_status: str, time_out = 'all' ,start_date: datetime = None, end_date: datetime = None) -> list[dict[Any, Any]]:
conn = DBService().get_conn()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO verify_log(inspectionAmount,cyjgxx,verify_time,file_path) values(?,?,?,?)",
(inspection_amount,cyjgxx,parser.parse(datetime.now().strftime("%Y-%m-%d %H:%M:%S")),file_path))
conn.commit()
sql = """
select id, checkCode, drawer, formType, invoiceAmountPreTax, invoiceCode,
invoiceDate, invoiceNumber, invoiceTax, invoiceType, machineCode,
passwordArea, printedInvoiceCode, printedInvoiceNumber,
purchaserBankAccountInfo, purchaserContactInfo, purchaserName,
purchaserTaxNumber, recipient, remarks, reviewer,
sellerBankAccountInfo, sellerContactInfo, sellerName, sellerTaxNumber,
specialTag, title, totalAmount, totalAmountInWords,file_path,verify_time,verify_status,desc,desc_files,status
from invoice where
"""
if time_out != 'all':
sql += " date('now', '-"+time_out+" day') >= date(verify_time) and verify_status == 'success' "
else:
sql += " verify_status = " + verify_status
if start_date is not None:
sql += " and verify_time >= '" + start_date.strftime("%Y-%m-%d") + "'"
if end_date is not None:
sql += " and verify_time <= '" + end_date.strftime("%Y-%m-%d") + "'"
if value != '':
sql += " and ( "
sql += "file_path like '%" + value + "%' or "
sql += "checkCode like '%" + value + "%' or "
sql += "drawer like '%" + value + "%' or "
sql += "formType like '%" + value + "%' or "
sql += "invoiceAmountPreTax like '%" + value + "%' or "
sql += "invoiceCode like '%" + value + "%' or "
sql += "invoiceDate like '%" + value + "%' or "
sql += "invoiceNumber like '%" + value + "%' or "
sql += "invoiceTax like '%" + value + "%' or "
sql += "invoiceType like '%" + value + "%' or "
sql += "machineCode like '%" + value + "%' or "
sql += "passwordArea like '%" + value + "%' or "
sql += "printedInvoiceCode like '%" + value + "%' or "
sql += "printedInvoiceNumber like '%" + value + "%' or "
sql += "purchaserBankAccountInfo like '%" + value + "%' or "
sql += "purchaserContactInfo like '%" + value + "%' or "
sql += "purchaserName like '%" + value + "%' or "
sql += "purchaserTaxNumber like '%" + value + "%' or "
sql += "recipient like '%" + value + "%' or "
sql += "remarks like '%" + value + "%' or "
sql += "reviewer like '%" + value + "%' or "
sql += "sellerBankAccountInfo like '%" + value + "%' or "
sql += "sellerContactInfo like '%" + value + "%' or "
sql += "sellerName like '%" + value + "%' or "
sql += "sellerTaxNumber like '%" + value + "%' or "
sql += "specialTag like '%" + value + "%' or "
sql += "title like '%" + value + "%' or "
sql += "totalAmount like '%" + value + "%' or "
sql += "totalAmountInWords like '%" + value + "%' or "
sql += "desc like '%" + value + "%' or "
sql += "desc_files like '%" + value + "%' "
sql += ")"
sql += " order by verify_time desc"
cursor.execute(sql)
rows = cursor.fetchall()
data = []
for row in rows:
data.append({
'checkCode': row[0],
'drawer': row[1],
'formType': row[2],
'invoiceAmountPreTax': row[3],
'invoiceCode': row[4],
'invoiceDate': row[5],
'invoiceNumber': row[6],
'invoiceTax': row[7],
'invoiceType': row[8],
'machineCode': row[9],
'passwordArea': row[10],
'printedInvoiceCode': row[11],
'printedInvoiceNumber': row[12],
'purchaserBankAccountInfo': row[13],
'purchaserContactInfo': row[14],
'purchaserName': row[15],
'purchaserTaxNumber': row[16],
'recipient': row[17],
'remarks': row[18],
'reviewer': row[19],
'sellerBankAccountInfo': row[20],
'sellerContactInfo': row[21],
'sellerName': row[22],
'sellerTaxNumber': row[23],
'specialTag': row[24],
'title': row[25],
'totalAmount': row[26],
'totalAmountInWords': row[27],
'file_path': row[28],
'verify_time': row[29],
'verify_status': row[30],
'desc': row[31],
'desc_files': row[32]
})
return data

5
alibabacloud_sample/dbservice.py
View File

@ -115,6 +115,11 @@ class DBService:
totalAmount REAL, -- 合计金额
totalAmountInWords TEXT, -- 大写金额
file_path TEXT, -- 文件哈希
verify_time TEXT, -- 验证时间
verify_status TEXT, -- 验证结果
desc TEXT, -- 备注
desc_files TEXT, -- 备注附件
status TEXT, -- 是否报销（yes,no）
UNIQUE (invoiceCode, invoiceNumber) -- 联合唯一约束
);
"""

15
alibabacloud_sample/fileservice.py
View File

@ -1,4 +1,6 @@
# 文件服务接口
from typing import Any
from flask import Request
from werkzeug.datastructures import FileStorage
from werkzeug.utils import secure_filename
@ -12,7 +14,7 @@ class FileService:
def __init__(self):
pass
# 上传文件
def upload(self, request: Request, file_path: str) -> str:
def upload(self, request: Request, file_path: str) -> None | str | dict[Any, Any] | Any:
if request.method == 'POST':
# 检查是否有文件在请求中
if 'file' not in request.files:
@ -29,9 +31,14 @@ class FileService:
file.save(filepath)
file_size = os.path.getsize(filepath)
file_hash = self.getHash(filepath)
vfile = dataservice.insert_file_data(file.filename, filepath, file_size, filetype, time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),file_hash)
print("insert vfile:", vfile)
return vfile
v_file = dataservice.get_file(file_hash,file_size)
if v_file:
os.remove(filepath)
return v_file
else:
v_file = dataservice.insert_file_data(file.filename, filepath, file_size, filetype, time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),file_hash)
print("insert v_file:", v_file)
return v_file
# 获取文件md5
def getHash(self, file_path: str) -> str:
with open(file_path, 'rb') as f:

49
alibabacloud_sample/main.py
View File

@ -10,6 +10,7 @@ app = Flask(__name__)
CORS(app)
app.config['UPLOAD_FOLDER'] = 'uploads' # 指定上传文件夹
app.config['WEBAPP'] = 'templates'
dataservice = DataService()
@app.route('/index')
def index():
return render_template('index.html')
@ -35,27 +36,57 @@ def upload_file():
def recognize():
file_path = request.args.get('filePath')
service = Service()
dataservice = DataService()
data = dataservice.get_invoice(file_path=file_path)
print(data)
if data is not None and data.get('file_path') is not None:
return data
return service.recognize(file_path=file_path)
# 发票验证
@app.route('/verify', methods=['GET'])
def verify():
file_path = request.args.get('filePath')
invoice_id = request.args.get('invoiceId')
if file_path is None:
return "请选择文件"
service = Service()
data = dataservice.get_invoice(file_path=file_path,invice_id=invoice_id)
if data is not None:
if data.get('status') == 'yes':
return "已报销发票！！"
return service.verify(data=data,file_path=file_path)
else:
return "请选择文件"
# 发票复验
@app.route('/reverify', methods=['GET'])
def reverify():
file_path = request.args.get('filePath')
if file_path is None:
return "请选择文件"
service = Service()
dataservice = DataService()
data = dataservice.get_invoice(file_path=file_path)
return service.verify(data=data,file_path=file_path)
@app.route('/list',methods=['POST'])
def list():
data = []
return data
return service.verify(data=data,file_path=file_path,reverify=True)
# 发票列表
@app.route('/listInvoice',methods=['POST'])
def listInvoice():
value = request.form.get('value')
verify_status = request.form.get('verifyStatus')
return dataservice.get_invoice_list(value=value,verify_status=verify_status)
@app.route('/listInvoiceTimeOut',methods=['POST'])
def listInvoiceTimeOut():
time_out = request.form.get('timeOut')
return dataservice.get_invoice_list(value='',verify_status='',time_out=time_out)
# 发票日志列表
@app.route('/listLogs',methods=['GET'])
def listLogs():
file_path = request.args.get('filePath')
return dataservice.get_verify_log_list(file_path=file_path)
# 更新发票标注
@app.route('/updateInvoice',methods=['POST'])
def updateInvoice():
invoice_id = request.form.get('invoiceId')
status = request.form.get('status')
desc = request.form.get('desc')
desc_files = request.form.get('desc_files')
return dataservice.update_invoice_desc(invoice_id=invoice_id,desc=desc,desc_file=desc_files,status=status)
if __name__ == '__main__':
# Service.verify(sys.argv[1:])
# Service.recognize(file_path="/home/jayus/图片/wechat_2025-07-31_131911_822.png")

34
alibabacloud_sample/service.py
View File

@ -66,12 +66,14 @@ class Service:
return ocr_api20210707Client(nconfig)
# 发票验证
@staticmethod
def verify(data: dict, file_path: str):
def verify(data: dict, file_path: str,reverify: bool = False):
dataservice = DataService()
data1 = dataservice.get_verify_log(file_path=file_path)
print(data)
if data1 is not None and data1.get('verify_time') is not None:
return data1
# 复验
if not reverify:
data1 = dataservice.get_verify_log(file_path=file_path)
print(data)
if data1 is not None and data1.get('inspectionAmount') > 0:
return data1
client = Service.create_client()
type = Service.typeDict.get(data.get('invoiceType'))
sum = data.get('totalAmount')
@ -94,11 +96,14 @@ class Service:
try:
resp = client.verify_vatinvoice_with_options(verify_vatinvoice_request, runtime)
ConsoleClient.log(UtilClient.to_jsonstring(resp))
data = json.loads(UtilClient.to_jsonstring(resp.body.data)).get('data')
print(data)
if data is not None:
dataservice.insert_verify_log(data.get('inspectionAmount'), data.get('cyjgxx'), file_path)
return data
resdata = json.loads(UtilClient.to_jsonstring(resp.body.data)).get('data')
if resdata is None:
resdata = {'cyjgxx': "未查询到发票信息", 'inspectionAmount': -1,'status': 'fail'}
else:
resdata.update({'status': 'success'})
dataservice.insert_verify_log(resdata.get('inspectionAmount'), resdata.get('cyjgxx'), file_path)
dataservice.update_invoice_status(data.get('id'), resdata.get('status'))
return resdata
except Exception as error:
# 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
# 错误 message
@ -121,7 +126,8 @@ class Service:
res = client.recognize_mixed_invoices_with_options(recognize_mixed_invoices_request, runtime)
print(UtilClient.to_jsonstring(res.body.data))
data = json.loads(UtilClient.to_jsonstring(res.body.data)).get('subMsgs')[0].get('result').get('data')
dataservice.insert_invoice_log(data, file_path)
data_id = dataservice.insert_invoice_log(data, file_path)
data.update({'id': data_id})
return data
except Exception as error:
# 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
@ -130,4 +136,8 @@ class Service:
# 诊断地址
print(error.data.get("Recommend"))
UtilClient.assert_as_string(error.message)
# 发票验证记录
@staticmethod
def get_verify_log_list(file_path):
dataservice = DataService()
return dataservice.get_verify_log_list(file_path)

BIN
alibabacloud_sample/verify.db Normal file
View File

Binary file not shown.