from reportlab.lib.pagesizes import A4 from reportlab.lib import colors from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle from reportlab.lib.units import inch, cm from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle from reportlab.lib.enums import TA_LEFT, TA_RIGHT, TA_CENTER from reportlab.pdfgen import canvas from reportlab.platypus import Image import streamlit as st import pandas as pd from io import BytesIO import xlsxwriter from datetime import date, datetime from st_aggrid import AgGrid from st_aggrid.grid_options_builder import GridOptionsBuilder from helpertools_TM import helpers class ReportGenerator: def __init__(self, processor, trova_errori, interface, no_zero=None): """Initialize ReportGenerator with required components. processor: Handles data processing and error collection trova_errori: Contains error detection logic and dataframes interface: Handles UI interactions and grid displays no_zero: Flag for handling zero values """ self.df = trova_errori.df self.df_bambini = trova_errori.df_bambini_che_impediscono_controllo_superamento_ore self.df_calcolo_666 = trova_errori.df_calcolo_666 self.dffinal = trova_errori.dffinal self.processor = processor self.no_zero = no_zero self.interface = interface self.trova_errori = trova_errori self.ERRORDICT = trova_errori.ERRORDICT self.helper = helpers(interface,self.ERRORDICT) def get_version(self): return "ReportGenerator v1.0.1" @st.fragment() def display_final_table(self) -> None: """ Displays the final table with all data and a separate table for errors only. This method creates two expanders: one for displaying all data in a grid format, and another for showing only records that contain errors. It also provides options to download these tables as Excel files. Args: None Returns: None """ # Display the final table with all data no_zero = self.no_zero # Assuming this is defined somewhere else in the code. # Create expander for full data table # This section shows all records, including those without errors expndr = st.expander("TABELLA FINALE ELABORATA - TUTTI I DATI") with expndr: # Enable Excel download of complete dataset # Sort by surname and name for better readability self.interface.download_excel_file( self.df.sort_values(by=["Cognome", "Nome"]), "TabellaFinaleElaborata_TuttiDati.xlsx", 'mantieniColonneErrore' ) # Configure and display interactive grid with all records # Enable enterprise features for better user interaction gridOptions = self.interface.buildGrid(self.df.sort_values(by=["Cognome", "Nome"])) AgGrid( self.df.sort_values(by=["Cognome", "Nome"]), gridOptions=gridOptions, enable_enterprise_modules=True, ) # Display the table with only errors # Create expander for error-only table # This section filters and shows only records with detected errors expndr = st.expander("TABELLA FINALE ELABORATA - SOLO ERRORI") with expndr: # Filter data to show only records with errors self.dffinal = self.helper.make_df_solo_errori(self.df) # Provide download option for the errors-only table self.interface.download_excel_file( self.dffinal.sort_values(by=["Cognome", "Nome"]), "TabellaFinaleElaborata_SoloErrori.xlsx", "mantieniColonneErrore" ) # Display the errors-only table in a grid format gridOptions = self.interface.buildGrid(self.dffinal.sort_values(by=["Cognome", "Nome"])) AgGrid( self.dffinal.sort_values(by=["Cognome", "Nome"]), gridOptions=gridOptions, enable_enterprise_modules=True, key="tabfinalex" ) # Save the errors-only table to an Excel file # try: # self.dffinal.to_excel("storico.xlsx") # st.success("Salvato file storico: OK") # except Exception as e: # st.error(f"ERRORE salvataggio file storico: errore({e})") def genera_report(self) -> None: """ Generates detailed reports in Excel and PDF formats and provides download options via Streamlit. This method filters errors, computes unique children and municipalities, and generates Excel and PDF reports if errors are present. It handles various exceptions that may occur during the report generation process. Args: None Returns: None """ # Generate detailed reports in Excel and PDF formats try: # Filter out specific errors and get current error state error_keys, errors_present = self._filter_errors(excluded_error="errMassimo543") # Early return if no checks are needed if not error_keys: st.warning("Nessun controllo da eseguire.") return # Calculate statistics about the dataset nr_bambini = self._compute_unique_children() nr_comuni = self._compute_unique_gemeinden() # Early return if no errors were found if not errors_present: st.info("Nessun errore rilevato nei controlli selezionati.") return # Generate reports and provide download options try: excel_data = self._generate_excel_report(nr_bambini, nr_comuni, errors_present) except (xlsxwriter.exceptions.FileCreateError, PermissionError) as e: st.error(f"File access error: {str(e)}") st.stop() except ValueError as ve: st.error(f"Invalid data format: {str(ve)}") st.stop() except KeyError as ke: st.error(f"Missing required column: {str(ke)}") st.stop() self._provide_downloads(excel_data) st.success("Generato file riassunto analisi") except KeyError as ke: # Handle missing or invalid data structure errors st.error(f"Chiave mancante o struttura dati non valida: {ke}") st.stop() except ValueError as ve: st.error(f"Problema di tipo o formato dati: {ve}") st.stop() except FileNotFoundError as fnf: st.error(f"File di riferimento non trovato: {fnf}") except PermissionError as pe: st.error(f"Permessi insufficienti per accedere al file: {pe}") except Exception as e: st.error(f"Errore imprevisto durante la generazione del report: {e}") st.stop() def _filter_errors(self, excluded_error: str): """ Excludes a specific error from the error dictionary and computes present errors. Args: excluded_error (str): The error key to be excluded from the error dictionary. Returns: Tuple[List[str], Dict[str, int]]: Filtered error keys and errors with counts > 0. """ # Exclude a specific error and compute present errors # Create list of error keys excluding specified error error_keys = [k for k in self.ERRORDICT.keys() if k != excluded_error] if not error_keys: return error_keys, {} # Calculate error occurrences and filter out zeros error_counts = self.df[error_keys].sum() errors_present = error_counts[error_counts > 0].to_dict() return error_keys, errors_present def _compute_unique_children(self) -> int: """ Computes the number of unique children based on 'Codice fiscale'. Args: None Returns: int: The number of unique children. """ # Compute the number of unique children return self.df["Codice fiscale"].nunique() def _compute_unique_gemeinden(self) -> int: """ Computes the number of unique municipalities based on 'Com_code'. Args: None Returns: int: The number of unique municipalities. """ # Compute the number of unique municipalities return self.df["Com_code"].nunique() def _generate_excel_report(self, nr_bambini: int, nr_comuni: int, errors_present: dict) -> bytes: """ Generates the Excel report and returns the binary data. Args: nr_bambini (int): Number of unique children. nr_comuni (int): Number of unique municipalities. errors_present (dict): Dictionary of errors with their counts. Returns: bytes: The binary data of the generated Excel report. """ # Generate the Excel report and return the binary data excel_buffer = BytesIO() with xlsxwriter.Workbook(excel_buffer) as workbook: worksheet = workbook.add_worksheet("Riassunto") try: self._write_excel_summary(worksheet, nr_bambini, nr_comuni, errors_present, workbook) except: st.error("Error during _write_excel_summary") try: self._write_excel_detailed_sections(worksheet, errors_present, workbook) except: st.error("Error during _write_excel_detailed") return excel_buffer.getvalue() def _write_excel_summary(self, worksheet, nr_bambini: int, nr_comuni: int, errors_present: dict, workbook): """ Writes the summary section of the Excel report. Args: worksheet: The worksheet object to write the summary to. nr_bambini (int): Number of unique children. nr_comuni (int): Number of unique municipalities. errors_present (dict): Dictionary of errors with their counts. workbook: The workbook object to format the cells. Returns: None """ # Write the summary section of the Excel report # Define cell formats for different sections bold_format = workbook.add_format({'bold': True, 'font_size': 12}) bold_format2 = workbook.add_format({'bold': True, 'font_size': 11}) header_format = workbook.add_format({'bold': True, 'bg_color': '#D7E4BC'}) title_format = workbook.add_format({'bold': True, 'font_size': 18}) row, col = 3, 0 # Starting from row 1 as row 0 is typically reserved # Write report title and basic statistics try: worksheet.write(row, col, "Riassunto analisi resoconti Tagesmuetter", title_format) worksheet.set_row(row, 30) # Set row height to 30 row += 4 except xlsxwriter.exceptions.XlsxWriterException as xlse: st.error(f"Worksheet write error: {str(xlse)}") raise # Re-raise to abort the process try: # Number of children worksheet.write(row, col, f"Sono stati elaborati i dati di {nr_bambini} bambini in {nr_comuni} comuni") row += 2 except: st.error("Error during Sono stati elaborati i dati di {nr_bambini} bambini in {nr_comuni} comuni") # Document all errors found in the analysis try: for error_key, count in errors_present.items(): if error_key not in self.df.columns: raise KeyError(f"Error column {error_key} not found in dataframe") if error_key not in self.ERRORDICT: raise ValueError(f"Undocumented error code: {error_key}") error_description = self.ERRORDICT[error_key] worksheet.write( row, col, f"Trovate {int(count)} occorrenze per il controllo {error_description}") row += 1 row += 2 # Blank row except: st.error("Error during Document all errors found in the analysis") # Document processing errors from file handling try: for key in self.processor.errori.keys(): #st.write(key) worksheet.write( row, col, f"Errori {key}:" + (" (file non usati)" if key == "critici" else ""), bold_format ) row += 1 for filename in self.processor.errori[key]: #st.write(filename) worksheet.write( row, col+1, filename.split('/', 1)[1] if '/' in filename else filename, #filename, #.split('/', 1)[1], ) for error in self.processor.errori[key][filename]: #st.write("entered error") #st.write(error) worksheet.write( row, col+2, error, ) row += 1 row += 1 row += 3 except Exception as e: st.error("Error during Document processing errors from file handling: " + str(e)) # Handle special cases: children without hour control try: if not self.df_bambini.empty: worksheet.write( row, col, "Bambini per i quali non e' possibile eseguire il controllo superamento ore nel periodo contrattuale", bold_format ) row += 1 # Write column headers for header_col, header in enumerate(self.df_bambini.columns): worksheet.write(row, header_col, header, header_format) row += 1 # Set fixed column widths worksheet.set_column(0, len(self.df_bambini.columns)-1, 25) # Write data rows for _, r in self.df_bambini.iterrows(): for header_col, header in enumerate(self.df_bambini.columns): value = r.get(header, "") if isinstance(value, pd.Timestamp): value = value.strftime("%d/%m/%Y") worksheet.write(row, header_col, value) row += 1 row += 3 # Blank rows except: st.error("Error during Handle special cases: children without hour control") # Document discrepancies in attachment 666 if not self.df_calcolo_666.empty: worksheet.set_column(0, len(self.df_calcolo_666.columns)-1, 25) filtered_df_ore = self.df_calcolo_666[self.df_calcolo_666['Confronto ore'] == False] filtered_df_bambini = self.df_calcolo_666[self.df_calcolo_666['Confronto bambini'] == False] if not filtered_df_ore.empty: worksheet.write( row, col, "Riscontrate incongruenze nel numero delle ore per l'allegato 666", bold_format) row += 2 # Define headers headers = ["Comune provenienza bambino", "Ore di servizio allegato 666", "Ore di assistenza report Comuni"] for i, header in enumerate(headers): worksheet.write(row, col + i + 1, header, header_format) row += 1 for _, r in filtered_df_ore.iterrows(): worksheet.write(row, col + 1, r.get("Comune provenienza bambino", "")) worksheet.write(row, col + 2, r.get("Ore di servizio allegato 666", "")) worksheet.write(row, col + 3, r.get("Ore di assistenza report Comuni", "")) row += 1 row +=1 # Blank rows if not filtered_df_bambini.empty: worksheet.write( row, col, "Riscontrate incongruenze nel numero bambini per l'allegato 666", bold_format) row += 2 # Define headers headers = ["Comune provenienza bambino", "Numero utenti allegato 666", "Numero bambini report Comuni"] for i, header in enumerate(headers): worksheet.write(row, col + i + 1, header, header_format) row += 1 for _, r in filtered_df_bambini.iterrows(): worksheet.write(row, col + 1, r.get("Comune provenienza bambino", "")) worksheet.write(row, col + 2, r.get("Numero utenti allegato 666", "")) worksheet.write(row, col + 3, r.get("Numero bambini report Comuni", "")) row += 1 def _write_excel_detailed_sections(self, worksheet, errors_present: dict, workbook): """ Writes detailed sections per error in the Excel report. Args: worksheet: The worksheet object to write the detailed sections to. errors_present (dict): Dictionary of errors with their counts. workbook: The workbook object to format the cells. Returns: None """ # Write detailed sections per error in the Excel report title_format2 = workbook.add_format({'bold': True, 'font_size': 12}) header_format = workbook.add_format({'bold': True, 'bg_color': '#D7E4BC'}) date_format = workbook.add_format({'num_format': 'dd/mm/yyyy'}) summary_format = workbook.add_format({'italic': False, 'font_size': 10}) row, col = worksheet.dim_rowmax, 0 row += 4 for error_key, count in errors_present.items(): error_description = self.ERRORDICT.get(error_key, error_key) worksheet.write(row, col, error_description, title_format2) row += 2 # Headers for detailed sections headers = [ "Cognome", "Nome", "Codice fiscale", "Data di nascita", "Comune", "Data inizio contratto", "Data fine contratto", ] for i, header in enumerate(headers): worksheet.write(row, col + i + 1, header, header_format) row += 1 for header_col, header in enumerate(headers, start=1): max_length = max( self.df[header].astype(str).map(len).max() if header in self.df.columns else 0, len(header) ) + 2 worksheet.set_column(col + header_col, col + header_col, max_length + 5) df_bambini = self.df[self.df[error_key] == True].sort_values(by=["Cognome", "Nome"]) for _, r in df_bambini.iterrows(): worksheet.write(row, col + 1, r.get("Cognome", "")) worksheet.write(row, col + 2, r.get("Nome", "")) worksheet.write(row, col + 3, r.get("Codice fiscale", "")) data_nascita = r.get("Data di nascita", "") if pd.notnull(data_nascita) and isinstance(data_nascita, pd.Timestamp): worksheet.write_datetime(row, col + 4, data_nascita, date_format) else: worksheet.write(row, col + 4, str(data_nascita)) worksheet.write(row, col + 5, r.get("Comune", "")) data_inizio = r.get("Data inizio contratto (o data inizio assistenza se diversa)", "") if pd.notnull(data_inizio) and isinstance(data_inizio, pd.Timestamp): worksheet.write_datetime(row, col + 6, data_inizio, date_format) else: worksheet.write(row, col + 6, str(data_inizio)) data_fine = r.get("Data fine contratto (o data fine assistenza se diversa)", "") if pd.notnull(data_fine) and isinstance(data_fine, pd.Timestamp): worksheet.write_datetime(row, col + 7, data_fine, date_format) else: worksheet.write(row, col + 7, str(data_fine)) row += 1 # Add summary line below the table row += 1 worksheet.write(row, col, "Riepilogo:", summary_format) #row += 1 # Collect all names and communes into a single summary line summary_entries = [] for _, r in df_bambini.iterrows(): # Capitalize each part of potentially compound names cognome_parts = [part.capitalize() for part in r.get("Cognome", "").split()] nome_parts = [part.capitalize() for part in r.get("Nome", "").split()] cognome = " ".join(cognome_parts).rstrip() nome = " ".join(nome_parts).rstrip() comune = r.get("Comune", "").rstrip() summary_entries.append(f"{cognome} {nome} ({comune})") # Join all entries with comma separators and write as a single line summary_line = ", ".join(summary_entries) worksheet.write(row, col + 1, summary_line, summary_format) row += 1 row += 3 def _prepare_additional_df(self) -> pd.DataFrame: """ Prepares the additional DataFrame for the PDF report by renaming and dropping columns. Args: None Returns: pd.DataFrame: The prepared additional DataFrame. """ # Prepare the additional DataFrame for the PDF report additional_df = self.df_bambini.copy() column_list = list(additional_df.columns) column_list[3] = 'Data Inizio Contratto' column_list[4] = 'Data Fine Contratto' additional_df.columns = column_list position_to_delete = 5 # Adjust based on actual column index if position_to_delete < len(additional_df.columns): column_name = additional_df.columns[position_to_delete] additional_df.drop(columns=column_name, inplace=True) return additional_df @st.fragment() def _provide_downloads(self, excel_data: bytes): #, pdf_data: bytes): """ Provides download buttons for the generated Excel and PDF reports. Args: excel_data (bytes): The binary data of the generated Excel report. # pdf_data (bytes): The binary data of the generated PDF report. Returns: None """ # Provide download buttons for the generated Excel and PDF reports timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") excel_file_name = f"riassuntoAutomatico_{timestamp}.xlsx" col1, col2 = st.columns(2) with col1: st.download_button( label="📥 Scarica report riassuntivo dell'analisi in formato Excel", data=excel_data, file_name=excel_file_name, mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", key="fileRiassuntoExcel", )