mirror of
https://github.com/Noettore/fattureCCSR.git
synced 2025-10-14 19:26:39 +02:00
fattureCCSR: changed log_dialog, without using logging
Signed-off-by: Ettore Dreucci <ettore.dreucci@gmail.com>
This commit is contained in:
@@ -9,15 +9,17 @@ import requests
|
||||
import requests_ntlm
|
||||
import openpyxl
|
||||
import PyPDF2
|
||||
import wx
|
||||
|
||||
import logger
|
||||
|
||||
def get_invoices_info(input_file_path: str) -> dict:
|
||||
def get_invoices_info(input_file_path: str) -> tuple:
|
||||
"""extract invoices IDs and URLs from xlsx input file"""
|
||||
xlsx_file = openpyxl.load_workbook(input_file_path)
|
||||
sheet = xlsx_file.active
|
||||
invoices = dict()
|
||||
|
||||
owner_name = '_'.join(sheet["B1"].value.split()[2:])
|
||||
print(owner_name)
|
||||
|
||||
for i in range(1, sheet.max_row+1):
|
||||
invoice_id = sheet["I"+str(i)].value
|
||||
if invoice_id is not None and "CCSR" in invoice_id:
|
||||
@@ -30,8 +32,8 @@ def get_invoices_info(input_file_path: str) -> dict:
|
||||
"good": None,
|
||||
}
|
||||
invoices[invoice_id] = invoice
|
||||
|
||||
return invoices
|
||||
invoices_info = (owner_name, invoices)
|
||||
return invoices_info
|
||||
|
||||
def open_file(file_path):
|
||||
"""open a file with the default software"""
|
||||
@@ -41,13 +43,15 @@ def open_file(file_path):
|
||||
opener = "open" if sys.platform == "darwin" else "xdg-open"
|
||||
subprocess.call([opener, file_path])
|
||||
|
||||
def download_invoices(input_file_path: str, output_file_path: str, username: str, password: str):
|
||||
def download_invoices(parent):
|
||||
"""download invoices from CCSR"""
|
||||
invoices = get_invoices_info(input_file_path)
|
||||
invoices_info = get_invoices_info(parent.input_file_path)
|
||||
invoices = invoices_info[1]
|
||||
|
||||
session = requests.Session()
|
||||
session.auth = requests_ntlm.HttpNtlmAuth("sr\\"+username, password)
|
||||
logger.downloader_logger.info("Inizio download fatture dal portale CCSR")
|
||||
session.auth = requests_ntlm.HttpNtlmAuth("sr\\"+parent.login_dlg.username.GetValue(), parent.login_dlg.password.GetValue())
|
||||
parent.log_dialog.log_text.AppendText("Inizio download fatture dal portale CCSR\n")
|
||||
wx.Yield()
|
||||
|
||||
tmp_dir = tempfile.mkdtemp()
|
||||
|
||||
@@ -60,28 +64,39 @@ def download_invoices(input_file_path: str, output_file_path: str, username: str
|
||||
with open(tmp_dir+"/"+invoice_id+".pdf", "wb") as output_file:
|
||||
output_file.write(resp.content)
|
||||
invoice["path"] = output_file.name
|
||||
print(invoice["path"])
|
||||
try:
|
||||
PyPDF2.PdfFileReader(open(invoice["path"], "rb"))
|
||||
except (PyPDF2.utils.PdfReadError, OSError):
|
||||
logger.downloader_logger.error("fattura %s corrotta!", invoice_id)
|
||||
parent.log_dialog.log_text.AppendText("Errore: fattura %s corrotta!\n" % invoice_id)
|
||||
wx.Yield()
|
||||
invoice["good"] = False
|
||||
else:
|
||||
downloaded_count += 1
|
||||
logger.downloader_logger.info("%d/%d scaricata fattura %s in %s", downloaded_count, invoices_count, invoice_id, invoice["path"])
|
||||
parent.log_dialog.log_text.AppendText("%d/%d scaricata fattura %s in %s\n" % (downloaded_count, invoices_count, invoice_id, invoice["path"]))
|
||||
wx.Yield()
|
||||
invoice["good"] = True
|
||||
else:
|
||||
logger.downloader_logger.error("impossibile scaricare fattura %s: %d", invoice_id, resp.status_code)
|
||||
parent.log_dialog.log_text.AppendText("Errore: impossibile scaricare fattura %s: %d\n" % (invoice_id, resp.status_code))
|
||||
wx.Yield()
|
||||
invoice["good"] = False
|
||||
|
||||
parent.output_pdf_dialog.SetFilename("fatture_%s.pdf" % invoices_info[0])
|
||||
|
||||
if parent.output_pdf_dialog.ShowModal() == wx.ID_OK:
|
||||
parent.output_file_path = parent.output_pdf_dialog.GetPath()
|
||||
else:
|
||||
#TODO: avviso errore file output
|
||||
return
|
||||
|
||||
merger = PyPDF2.PdfFileMerger()
|
||||
for invoice in invoices.values():
|
||||
if invoice["good"]:
|
||||
merger.append(PyPDF2.PdfFileReader(open(invoice["path"], "rb")))
|
||||
merger.write(output_file_path)
|
||||
merger.write(parent.output_file_path)
|
||||
|
||||
open_file(output_file_path)
|
||||
open_file(parent.output_file_path)
|
||||
|
||||
shutil.rmtree(tmp_dir, ignore_errors=True)
|
||||
|
||||
logger.downloader_logger.info("Download terminato. Il pdf contenente le fatture si trova in %s", output_file_path)
|
||||
parent.log_dialog.log_text.AppendText("Download terminato. Il pdf contenente le fatture si trova in %s\n" % parent.output_file_path)
|
||||
wx.Yield()
|
||||
|
@@ -1,56 +1,31 @@
|
||||
"""This utility is used for downloading or converting to TRAF2000 invoices from a .csv or .xml report file"""
|
||||
|
||||
import logging
|
||||
import wx
|
||||
|
||||
import downloader
|
||||
import traf2000_converter
|
||||
import exc
|
||||
import utils
|
||||
import logger
|
||||
|
||||
DOWNLOAD_ACTION = 1
|
||||
CONVERT_ACTION = 2
|
||||
|
||||
class LogHandler(logging.StreamHandler):
|
||||
"""logging stream handler"""
|
||||
def __init__(self, textctrl):
|
||||
logging.StreamHandler.__init__(self)
|
||||
self.textctrl = textctrl
|
||||
|
||||
def emit(self, record):
|
||||
"""constructor"""
|
||||
msg = self.format(record)
|
||||
self.textctrl.WriteText(msg + "\n")
|
||||
self.flush()
|
||||
|
||||
class LogDialog(wx.Dialog):
|
||||
"""logging panel"""
|
||||
def __init__(self, parent, title, action):
|
||||
super(LogDialog, self).__init__(parent, wx.ID_ANY, title)
|
||||
if action == DOWNLOAD_ACTION:
|
||||
self.logger = logger.downloader_logger
|
||||
elif action == CONVERT_ACTION:
|
||||
self.logger = logger.converter_logger
|
||||
else:
|
||||
raise exc.InvalidActionError(action)
|
||||
|
||||
log_text = wx.TextCtrl(self, wx.ID_ANY, size=(300, 200), style=wx.TE_MULTILINE|wx.TE_READONLY|wx.HSCROLL)
|
||||
log_handler = LogHandler(log_text)
|
||||
log_handler.setLevel(logging.INFO)
|
||||
self.logger.addHandler(log_handler)
|
||||
|
||||
main_sizer = wx.BoxSizer(wx.VERTICAL)
|
||||
|
||||
self.log_text = wx.TextCtrl(self, wx.ID_ANY, size=(300, 200), style=wx.TE_MULTILINE|wx.TE_READONLY|wx.HSCROLL)
|
||||
log_sizer = wx.BoxSizer(wx.HORIZONTAL)
|
||||
log_sizer.Add(log_text, 0, wx.ALL, 2)
|
||||
log_sizer.Add(self.log_text, 0, wx.ALL, 2)
|
||||
|
||||
self.log_text.Bind(wx.EVT_TEXT, self.on_text_update)
|
||||
|
||||
if action == CONVERT_ACTION:
|
||||
self.nc_logger = logger.note_credito_logger
|
||||
nc_text = wx.TextCtrl(self, wx.ID_ANY, size=(300, 200), style=wx.TE_MULTILINE|wx.TE_READONLY|wx.HSCROLL)
|
||||
nc_handler = LogHandler(nc_text)
|
||||
self.nc_logger.addHandler(nc_handler)
|
||||
|
||||
log_sizer.Add(nc_text, 0, wx.ALL, 2)
|
||||
self.nc_text = wx.TextCtrl(self, wx.ID_ANY, size=(300, 200), style=wx.TE_MULTILINE|wx.TE_READONLY|wx.HSCROLL)
|
||||
log_sizer.Add(self.nc_text, 0, wx.ALL, 2)
|
||||
|
||||
main_sizer.Add(log_sizer, 0, wx.ALL, 2)
|
||||
self.btn = wx.Button(self, wx.ID_OK, "Chiudi")
|
||||
@@ -59,6 +34,11 @@ class LogDialog(wx.Dialog):
|
||||
|
||||
self.SetSizerAndFit(main_sizer)
|
||||
|
||||
def on_text_update(self, event):
|
||||
"""autoscroll on text update"""
|
||||
self.ScrollPages(-1)
|
||||
event.Skip()
|
||||
|
||||
class LoginDialog(wx.Dialog):
|
||||
"""login dialog for basic auth download"""
|
||||
def __init__(self, parent, title):
|
||||
@@ -158,25 +138,23 @@ class FattureCCSRFrame(wx.Frame):
|
||||
print(handled_exception.args[0])
|
||||
if self.input_file_ext == ".xlsx":
|
||||
self.download_btn.Enable()
|
||||
self.traf2000_btn.Disable()
|
||||
elif self.input_file_ext in (".csv", ".xml"):
|
||||
self.traf2000_btn.Enable()
|
||||
self.download_btn.Disable()
|
||||
else:
|
||||
self.download_btn.Disable()
|
||||
self.traf2000_btn.Disable()
|
||||
|
||||
def btn_onclick(self, event):
|
||||
"""event raised when a button is clicked"""
|
||||
btn_id = event.GetEventObject().GetId()
|
||||
if btn_id == DOWNLOAD_ACTION:
|
||||
if self.output_pdf_dialog.ShowModal() == wx.ID_OK:
|
||||
self.output_file_path = self.output_pdf_dialog.GetPath()
|
||||
else:
|
||||
#TODO: avviso errore file output
|
||||
return
|
||||
self.login_dlg.ShowModal()
|
||||
if self.login_dlg.logged_id:
|
||||
self.log_dialog = LogDialog(self, "Download delle fatture dal portale CCSR", DOWNLOAD_ACTION)
|
||||
self.log_dialog.Show()
|
||||
downloader.download_invoices(self.input_file_path, self.output_file_path, self.login_dlg.username.GetValue(), self.login_dlg.password.GetValue())
|
||||
downloader.download_invoices(self)
|
||||
self.log_dialog.btn.Enable()
|
||||
elif btn_id == CONVERT_ACTION:
|
||||
if self.output_traf2000_dialog.ShowModal() == wx.ID_OK:
|
||||
@@ -188,7 +166,7 @@ class FattureCCSRFrame(wx.Frame):
|
||||
self.log_dialog.Show()
|
||||
#TODO: error frame
|
||||
try:
|
||||
traf2000_converter.convert(self.input_file_path, self.output_file_path)
|
||||
traf2000_converter.convert(self.input_file_path, self.output_file_path, self)
|
||||
except exc.NoFileError as handled_exception:
|
||||
print(handled_exception.args[0])
|
||||
except exc.NoFileExtensionError as handled_exception:
|
||||
|
@@ -4,11 +4,11 @@ import datetime
|
||||
import csv
|
||||
import xml.etree.ElementTree
|
||||
import unidecode
|
||||
import wx
|
||||
|
||||
import utils
|
||||
import logger
|
||||
|
||||
def import_csv(csv_file_path: str) -> dict:
|
||||
def import_csv(csv_file_path: str, parent) -> dict:
|
||||
"""Return a dict containing the invoices info"""
|
||||
fatture = dict()
|
||||
with open(csv_file_path, newline="") as csv_file:
|
||||
@@ -33,7 +33,7 @@ def import_csv(csv_file_path: str) -> dict:
|
||||
"tipoFattura": tipo_fattura,
|
||||
"rifFattura": linea[4],
|
||||
"dataFattura": linea[2].replace("/", ""),
|
||||
"ragioneSociale": unidecode.unidecode(linea[6] + " " + " ".join(linea[5].split(" ")[0:2])),
|
||||
"ragioneSociale": unidecode.unidecode(linea[6] + " " + " ".join(linea[5].split()[0:2])),
|
||||
"posDivide": str(len(linea[6]) + 1),
|
||||
"cf": linea[7],
|
||||
"importoTotale": 0,
|
||||
@@ -48,10 +48,11 @@ def import_csv(csv_file_path: str) -> dict:
|
||||
else:
|
||||
fatture[num_fattura]["importoTotale"] += importo
|
||||
fatture[num_fattura]["righe"][linea[14]] = importo
|
||||
logger.converter_logger.info("Importata fattura n. %s", num_fattura)
|
||||
parent.log_dialog.log_text.AppendText("Importata fattura n. %s\n" % num_fattura)
|
||||
wx.Yield()
|
||||
return fatture
|
||||
|
||||
def import_xml(xml_file_path: str) -> dict:
|
||||
def import_xml(xml_file_path: str, parent) -> dict:
|
||||
"""Return a dict containing the invoices info"""
|
||||
fatture = dict()
|
||||
|
||||
@@ -82,7 +83,7 @@ def import_xml(xml_file_path: str) -> dict:
|
||||
"tipoFattura": tipo_fattura,
|
||||
"rifFattura": fattura.get('protocollo_fatturatestata1'),
|
||||
"dataFattura": datetime.datetime.fromisoformat(fattura.get('data_fatturatestata')).strftime("%d%m%Y"),
|
||||
"ragioneSociale": unidecode.unidecode(fattura.get('cognome_cliente') + ' ' + ' '.join(fattura.get('nome_cliente').split(' ')[0:2])),
|
||||
"ragioneSociale": unidecode.unidecode(fattura.get('cognome_cliente') + ' ' + ' '.join(fattura.get('nome_cliente').split()[0:2])),
|
||||
"posDivide": str(len(fattura.get('cognome_cliente')) + 1),
|
||||
"cf": fattura.get('cf_piva_cliente'),
|
||||
"importoTotale": importo_totale,
|
||||
@@ -90,35 +91,40 @@ def import_xml(xml_file_path: str) -> dict:
|
||||
"righe": righe,
|
||||
}
|
||||
fatture[num_fattura] = fattura_elem
|
||||
logger.converter_logger.info("Importata fattura n. %s", num_fattura)
|
||||
parent.log_dialog.log_text.AppendText("Importata fattura n. %s\n" % num_fattura)
|
||||
wx.Yield()
|
||||
return fatture
|
||||
|
||||
|
||||
def convert(input_file_path: str, out_file_path: str):
|
||||
def convert(input_file_path: str, out_file_path: str, parent):
|
||||
"""Output to a file the TRAF2000 records"""
|
||||
|
||||
input_file_ext = utils.file_extension(input_file_path, (".xml", ".csv"))
|
||||
if input_file_ext == ".csv":
|
||||
fatture = import_csv(input_file_path)
|
||||
fatture = import_csv(input_file_path, parent)
|
||||
|
||||
elif input_file_ext == ".xml":
|
||||
fatture = import_xml(input_file_path)
|
||||
fatture = import_xml(input_file_path, parent)
|
||||
|
||||
with open(out_file_path, "w") as traf2000_file:
|
||||
logger.note_credito_logger.info("Note di credito:")
|
||||
parent.log_dialog.nc_text.AppendText("Note di credito:\n")
|
||||
wx.Yield()
|
||||
|
||||
for fattura in fatture.values():
|
||||
if fattura["tipoFattura"] != "Fattura" and fattura["tipoFattura"] != "Nota di credito":
|
||||
logger.converter_logger.error("Errore: il documento %s può essere FATTURA o NOTA DI CREDITO", fattura["numFattura"])
|
||||
parent.log_dialog.log_text.AppendText("Errore: il documento %s può essere FATTURA o NOTA DI CREDITO\n" % fattura["numFattura"])
|
||||
wx.Yield()
|
||||
continue
|
||||
|
||||
if len(fattura["cf"]) != 16 and len(fattura["cf"]) == 11:
|
||||
logger.converter_logger.error("Errore: il documento %s non ha cf/piva", fattura["numFattura"])
|
||||
parent.log_dialog.log_text.AppendText("Errore: il documento %s non ha cf/piva\n" % fattura["numFattura"])
|
||||
wx.Yield()
|
||||
continue
|
||||
|
||||
if fattura["tipoFattura"] == "Nota di credito":
|
||||
# As for now this script doesn't handle "Note di credito"
|
||||
logger.note_credito_logger.info(fattura["numFattura"])
|
||||
parent.log_dialog.nc_text.AppendText(fattura["numFattura"]+"\n")
|
||||
wx.Yield()
|
||||
continue
|
||||
|
||||
linea = ["04103", "3", "0", "00000"] # TRF-DITTA + TRF-VERSIONE + TRF-TARC + TRF-COD-CLIFOR
|
||||
@@ -236,7 +242,8 @@ def convert(input_file_path: str, out_file_path: str):
|
||||
linea.append('S') # TRF-RIF-FATTURA
|
||||
linea.append('S' + ' '*2 + 'S' + ' '*2) # TRF-RISERVATO-B + TRF-MASTRO-CF + TRF-MOV-PRIVATO + TRF-SPESE-MEDICHE + TRF-FILLER
|
||||
linea.append('\n')
|
||||
logger.converter_logger.info("Creato record #0 per fattura n. %s", fattura["numFattura"])
|
||||
parent.log_dialog.log_text.AppendText("Creato record #0 per fattura n. %s\n" % fattura["numFattura"])
|
||||
wx.Yield()
|
||||
|
||||
#RECORD 5 per Tessera Sanitaria
|
||||
linea.append('04103' + '3' + '5') # TRF5-DITTA + TRF5-VERSIONE + TRF5-TARC
|
||||
@@ -256,7 +263,8 @@ def convert(input_file_path: str, out_file_path: str):
|
||||
linea.append((' ' + ' ' + ' ')*49) # TRF-A21CO-TIPO + TRF-A21CO-TIPO-SPESA + TRF-A21CO-FLAG-SPESA
|
||||
linea.append(' ' + 'S' + ' '*76) # TRF-SPESE-FUNEBRI + TRF-A21CO-PAGAM + FILLER + FILLER
|
||||
linea.append('\n')
|
||||
logger.converter_logger.info("Creato record #5 per fattura n. %s", fattura["numFattura"])
|
||||
parent.log_dialog.log_text.AppendText("Creato record #5 per fattura n. %s\n" % fattura["numFattura"])
|
||||
wx.Yield()
|
||||
|
||||
#RECORD 1 per num. doc. originale
|
||||
linea.append('04103' + '3' + '1') # TRF1-DITTA + TRF1-VERSIONE + TRF1-TARC
|
||||
@@ -275,10 +283,13 @@ def convert(input_file_path: str, out_file_path: str):
|
||||
linea.append(' '*8) # TRF-CK-RCHARGE
|
||||
linea.append('0'*(15-len(fattura["numFattura"])) + fattura["numFattura"]) # TRF-XNUM-DOC-ORI
|
||||
linea.append(' ' + '00' + ' '*1090) # TRF-MEM-ESIGIB-IVA + TRF-COD-IDENTIFICATIVO + TRF-ID-IMPORTAZIONE + TRF-XNUM-DOC-ORI-20 + SPAZIO + FILLER
|
||||
logger.converter_logger.info("Creato record #1 per fattura n. %s", fattura["numFattura"])
|
||||
parent.log_dialog.log_text.AppendText("Creato record #1 per fattura n. %s\n" % fattura["numFattura"])
|
||||
wx.Yield()
|
||||
|
||||
linea = ''.join(linea) + '\n'
|
||||
|
||||
traf2000_file.write(linea)
|
||||
logger.converter_logger.info("Convertita fattura n. %s", fattura["numFattura"])
|
||||
logger.converter_logger.info("Conversione terminata")
|
||||
parent.log_dialog.log_text.AppendText("Convertita fattura n. %s\n" % fattura["numFattura"])
|
||||
wx.Yield()
|
||||
parent.log_dialog.log_text.AppendText("Conversione terminata")
|
||||
wx.Yield()
|
||||
|
Reference in New Issue
Block a user