122 lines
5.3 KiB
Python
122 lines
5.3 KiB
Python
# 数据管理服务类
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from dateutil import parser
|
|
|
|
from alibabacloud_sample.dbservice import DBService
|
|
|
|
|
|
class DataService:
|
|
conn = None
|
|
def __init__(self):
|
|
self.conn = DBService().get_conn()
|
|
# 插入文件记录
|
|
def insert_file_data(self, file_name: str, file_path: str, file_size: int, file_type: str,file_time: str, file_hash: str):
|
|
conn = DBService().get_conn()
|
|
cursor = conn.cursor()
|
|
try:
|
|
cursor.execute(
|
|
"INSERT INTO file_data(file_name,file_path,file_size,file_type,file_time,file_hash) values(?,?,?,?,?,?)",
|
|
(file_name, file_path, file_size, file_type, file_time, file_hash))
|
|
conn.commit()
|
|
except Exception as e:
|
|
print(e)
|
|
conn.rollback()
|
|
cursor.execute("select file_path from file_data where file_hash = ?", (file_hash,))
|
|
rows = cursor.fetchall()
|
|
return rows[0][0]
|
|
return file_path
|
|
# 根据文件路径获取hash
|
|
def get_invoice(self, file_path: str) -> dict[Any, Any]:
|
|
conn = DBService().get_conn()
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
select 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 = ?
|
|
"""
|
|
, (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_invoice_log(self, data: dict, file_path: str):
|
|
conn = DBService().get_conn()
|
|
cursor = conn.cursor()
|
|
# 插入主表数据
|
|
invoice_insert = """
|
|
INSERT INTO invoice (
|
|
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
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
"""
|
|
invoice_values = (
|
|
data.get('checkCode', ''),
|
|
data.get('drawer', ''),
|
|
data.get('formType', ''),
|
|
float(data.get('invoiceAmountPreTax', 0.0)),
|
|
data.get('invoiceCode', ''),
|
|
data.get('invoiceDate', ''),
|
|
data.get('invoiceNumber', ''),
|
|
float(data.get('invoiceTax', 0.0)),
|
|
data.get('invoiceType', ''),
|
|
data.get('machineCode', ''),
|
|
data.get('passwordArea', ''),
|
|
data.get('printedInvoiceCode', ''),
|
|
data.get('printedInvoiceNumber', ''),
|
|
data.get('purchaserBankAccountInfo', ''),
|
|
data.get('purchaserContactInfo', ''),
|
|
data.get('purchaserName', ''),
|
|
data.get('purchaserTaxNumber', ''),
|
|
data.get('recipient', ''),
|
|
data.get('remarks', ''),
|
|
data.get('reviewer', ''),
|
|
data.get('sellerBankAccountInfo', ''),
|
|
data.get('sellerContactInfo', ''),
|
|
data.get('sellerName', ''),
|
|
data.get('sellerTaxNumber', ''),
|
|
data.get('specialTag', ''),
|
|
data.get('title', ''),
|
|
float(data.get('totalAmount', 0.0)),
|
|
data.get('totalAmountInWords', ''),
|
|
file_path
|
|
)
|
|
cursor.execute(invoice_insert, invoice_values)
|
|
conn.commit()
|
|
invoice_id = cursor.lastrowid
|
|
return invoice_id
|
|
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 = ?
|
|
"""
|
|
, (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() |