Download src/reportmaker.py from eternaut/RobertaB: direct link, hf CLI and curl.
- Browser
- Download file 23.6 kB
-
https://huggingface.co/spaces/eternaut/RobertaB/resolve/main/src/reportmaker.py
- Command line
-
hf download hf://spaces/eternaut/RobertaB/src/reportmaker.py
-
curl -L -o reportmaker.py https://huggingface.co/spaces/eternaut/RobertaB/resolve/main/src/reportmaker.py
23.6 kB
| 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" | |
| 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 | |
| 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", | |
| ) | |