Spaces:
Running
Running
| import gradio as gr | |
| import pandas as pd | |
| import os | |
| import zipfile | |
| from io import BytesIO | |
| import time | |
| import logging | |
| import datetime # Add this import | |
| # Configure logging | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format='%(asctime)s - %(levelname)s - %(message)s', | |
| handlers=[ | |
| logging.FileHandler('statistical_processor.log'), | |
| logging.StreamHandler() | |
| ] | |
| ) | |
| logger = logging.getLogger(__name__) | |
| # Create output directory if it doesn't exist | |
| os.makedirs("output", exist_ok=True) | |
| logger.info("Output directory checked/created") | |
| class ExcelDataProcessor: | |
| def __init__(self): | |
| self.output_files = [] | |
| logger.info("ExcelDataProcessor initialized") | |
| def _extract_number(self, sheet_name): | |
| """Extract number from sheet name for sorting""" | |
| try: | |
| return int(sheet_name[:sheet_name.find('.')]) | |
| except ValueError: | |
| logger.info(f"Error processing {sheet_name}: '>' not supported between instances of 'datetime.date' and 'str'") | |
| return float('inf') # Send sheets without numbers to the end | |
| def _create_unique_filename(self, base_name): | |
| """Create a unique filename with timestamp""" | |
| timestamp = int(time.time()) | |
| return f"{base_name}_{timestamp}.xlsx" | |
| def count_daily_registers_by_source_name(self, df): | |
| """Count daily registers by source name""" | |
| logger.info("Starting count_daily_registers_by_source_name") | |
| df_filtered = df.copy() | |
| df_filtered.loc[:, 'Created At'] = pd.to_datetime(df_filtered['Created At']).dt.date | |
| target_date = datetime.date(2025, 4, 14) | |
| df_filtered = df_filtered[df_filtered['Created At'] > target_date].copy() | |
| pivot_table = pd.pivot_table( | |
| df_filtered, | |
| index='Source Name', | |
| columns='Created At', | |
| values='User ID', | |
| aggfunc='count', | |
| fill_value=0 | |
| ) | |
| pivot_table.loc['Total'] = pivot_table.sum() | |
| output_path = self._create_unique_filename("count_daily_registers_by_source_name") | |
| full_path = os.path.join("output", output_path) | |
| pivot_table.to_excel(full_path) | |
| self.output_files.append(full_path) | |
| logger.info(f"Saved count_daily_registers_by_source_name to {full_path}") | |
| return pivot_table | |
| def count_daily_registers_by_ref(self, df): | |
| """Count daily registers by reference""" | |
| logger.info("Starting count_daily_registers_by_ref") | |
| df_filtered = df.copy() | |
| df_filtered.loc[:, 'Created At'] = pd.to_datetime(df_filtered['Created At']).dt.date | |
| target_date = datetime.date(2025, 4, 15) | |
| df_filtered = df_filtered[df_filtered['Created At'] < target_date].copy() | |
| df_filtered.loc[(df_filtered['Source Name'] == 'direct') & (df_filtered['Ref By'].isna()), 'Ref By'] = 'direct' | |
| pivot_table = pd.pivot_table( | |
| df_filtered, | |
| index='Ref By', | |
| columns='Created At', | |
| values='User ID', | |
| aggfunc='count', | |
| fill_value=0 | |
| ) | |
| pivot_table.loc['Total'] = pivot_table.sum() | |
| output_path = self._create_unique_filename("count_daily_registers_by_ref") | |
| full_path = os.path.join("output", output_path) | |
| pivot_table.to_excel(full_path) | |
| self.output_files.append(full_path) | |
| logger.info(f"Saved count_daily_registers_by_ref to {full_path}") | |
| return pivot_table | |
| def count_users_by_source_name(self, df): | |
| """Count unique users by source name""" | |
| logger.info("Starting count_users_by_source_name") | |
| df_filtered = df.copy() | |
| df_filtered = df_filtered.drop_duplicates(subset=['User ID'], keep='first') | |
| target_date = datetime.date(2025, 4, 14) | |
| df_filtered['Created At'] = pd.to_datetime(df_filtered['Created At']).dt.date | |
| df_filtered = df_filtered[df_filtered['Created At'] > target_date].copy() | |
| pivot_table = pd.pivot_table( | |
| df_filtered, | |
| index='Source Name', | |
| values='User ID', | |
| aggfunc='count', | |
| fill_value=0 | |
| ) | |
| output_path = self._create_unique_filename("count_users_by_source_name") | |
| full_path = os.path.join("output", output_path) | |
| pivot_table.to_excel(full_path) | |
| self.output_files.append(full_path) | |
| logger.info(f"Saved count_users_by_source_name to {full_path}") | |
| return pivot_table | |
| def count_users_by_ref(self, df): | |
| """Count unique users by reference""" | |
| logger.info("Starting count_users_by_ref") | |
| df_filtered = df.copy() | |
| df_filtered = df_filtered.drop_duplicates(subset=['User ID'], keep='first') | |
| target_date = datetime.date(2025, 4, 15) | |
| df_filtered['Created At'] = pd.to_datetime(df_filtered['Created At']).dt.date | |
| df_filtered = df_filtered[df_filtered['Created At'] < target_date].copy() | |
| df_filtered.loc[(df_filtered['Source Name'] == 'direct') & (df_filtered['Ref By'].isna()), 'Ref By'] = 'direct' | |
| pivot_table = pd.pivot_table( | |
| df_filtered, | |
| index='Ref By', | |
| values='User ID', | |
| aggfunc='count', | |
| fill_value=0 | |
| ) | |
| output_path = self._create_unique_filename("count_users_by_ref") | |
| full_path = os.path.join("output", output_path) | |
| pivot_table.to_excel(full_path) | |
| self.output_files.append(full_path) | |
| logger.info(f"Saved count_users_by_ref to {full_path}") | |
| return pivot_table | |
| def count_users_each_sheet_by_source_name(self, excel_file): | |
| """Count users in each sheet by source name""" | |
| logger.info("Starting count_users_each_sheet_by_source_name") | |
| df_dict = pd.read_excel(excel_file, sheet_name=None) | |
| sheet_dfs = {sheet_name: df for sheet_name, df in df_dict.items() if '.' in sheet_name} | |
| if not sheet_dfs: | |
| logger.warning("No sheets found with '.' in their names") | |
| return "No sheets found with '.' in their names", None | |
| combined_df = pd.concat( | |
| [df.assign(SheetName=sheet_name) for sheet_name, df in sheet_dfs.items()], | |
| axis=0, | |
| ignore_index=True | |
| ) | |
| combined_df = combined_df.dropna(how='all').copy() | |
| combined_df_filtered = combined_df.dropna(subset=['Source Name']).copy() | |
| combined_df_filtered['Created At'] = pd.to_datetime(combined_df_filtered['Created At']).dt.date | |
| target_date = datetime.date(2025, 4, 14) | |
| combined_df_filtered = combined_df_filtered[combined_df_filtered['Created At'] > target_date].copy() | |
| if not {'Source Name', 'User ID', 'SheetName'}.issubset(combined_df_filtered.columns): | |
| return "Required columns 'Source Name', 'User ID', or 'SheetName' not found", None | |
| pivot_table = pd.pivot_table( | |
| combined_df_filtered, | |
| index='Source Name', | |
| columns='SheetName', | |
| values='User ID', | |
| aggfunc='count', | |
| fill_value=0 | |
| ) | |
| sorted_columns = sorted(pivot_table.columns, key=self._extract_number) | |
| pivot_table = pivot_table[sorted_columns] | |
| pivot_table.loc['Total'] = pivot_table.sum() | |
| output_path = self._create_unique_filename("count_users_each_sheet_by_source_name") | |
| full_path = os.path.join("output", output_path) | |
| pivot_table.to_excel(full_path) | |
| self.output_files.append(full_path) | |
| logger.info(f"Saved count_users_each_sheet_by_source_name to {full_path}") | |
| return "Success", pivot_table | |
| def count_users_each_sheet_by_ref(self, excel_file): | |
| """Count users in each sheet by reference""" | |
| logger.info("Starting count_users_each_sheet_by_ref") | |
| df_dict = pd.read_excel(excel_file, sheet_name=None) | |
| sheet_dfs = {sheet_name: df for sheet_name, df in df_dict.items() if '.' in sheet_name} | |
| if not sheet_dfs: | |
| logger.warning("No sheets found with '.' in their names") | |
| return "No sheets found with '.' in their names", None | |
| combined_df = pd.concat( | |
| [df.assign(SheetName=sheet_name) for sheet_name, df in sheet_dfs.items()], | |
| axis=0, | |
| ignore_index=True | |
| ) | |
| combined_df = combined_df.dropna(how='all').copy() | |
| combined_df_filtered = combined_df.copy() | |
| combined_df_filtered['Created At'] = pd.to_datetime(combined_df_filtered['Created At']).dt.date | |
| target_date = datetime.date(2025, 4, 15) | |
| combined_df_filtered = combined_df_filtered[combined_df_filtered['Created At'] < target_date].copy() | |
| combined_df_filtered.loc[(combined_df_filtered['Source Name'] == 'direct') & (combined_df_filtered['Ref By'].isna()), 'Ref By'] = 'direct' | |
| if not {'Ref By', 'User ID', 'SheetName'}.issubset(combined_df_filtered.columns): | |
| return "Required columns 'Ref By', 'User ID', or 'SheetName' not found", None | |
| pivot_table = pd.pivot_table( | |
| combined_df_filtered, | |
| index='Ref By', | |
| columns='SheetName', | |
| values='User ID', | |
| aggfunc='count', | |
| fill_value=0 | |
| ) | |
| sorted_columns = sorted(pivot_table.columns, key=self._extract_number) | |
| pivot_table = pivot_table[sorted_columns] | |
| pivot_table.loc['Total'] = pivot_table.sum() | |
| output_path = self._create_unique_filename("count_users_each_sheet_by_ref") | |
| full_path = os.path.join("output", output_path) | |
| pivot_table.to_excel(full_path) | |
| self.output_files.append(full_path) | |
| logger.info(f"Saved count_users_each_sheet_by_ref to {full_path}") | |
| return "Success", pivot_table | |
| def count_users_each_sheet_by_date(self, excel_file): | |
| """Count users in each sheet by date""" | |
| logger.info("Starting count_users_each_sheet_by_date") | |
| df_dict = pd.read_excel(excel_file, sheet_name=None) | |
| sheet_dfs = {sheet_name: df for sheet_name, df in df_dict.items() if '.' in sheet_name} | |
| if not sheet_dfs: | |
| logger.warning("No sheets found with '.' in their names") | |
| return "No sheets found with '.' in their names", None | |
| combined_df = pd.concat( | |
| [df.assign(SheetName=sheet_name) for sheet_name, df in sheet_dfs.items()], | |
| axis=0, | |
| ignore_index=True | |
| ) | |
| combined_df = combined_df.dropna(how='all').copy() | |
| combined_df_filtered = combined_df[combined_df['Created At'].notna()].copy() | |
| combined_df_filtered.loc[:, 'Created At'] = pd.to_datetime(combined_df_filtered['Created At']).dt.date | |
| if not {'Created At', 'User ID', 'SheetName'}.issubset(combined_df_filtered.columns): | |
| return "Required columns 'Created At', 'User ID', or 'SheetName' not found", None | |
| pivot_table = pd.pivot_table( | |
| combined_df_filtered, | |
| index='Created At', | |
| columns='SheetName', | |
| values='User ID', | |
| aggfunc='count', | |
| fill_value=0 | |
| ) | |
| sorted_columns = sorted(pivot_table.columns, key=self._extract_number) | |
| pivot_table = pivot_table[sorted_columns] | |
| pivot_table.loc['Total'] = pivot_table.sum() | |
| output_path = self._create_unique_filename("count_users_each_sheet_by_date") | |
| full_path = os.path.join("output", output_path) | |
| pivot_table.to_excel(full_path) | |
| self.output_files.append(full_path) | |
| logger.info(f"Saved count_users_each_sheet_by_date to {full_path}") | |
| return "Success", pivot_table | |
| def process_file(self, excel_file, operations): | |
| """Process file with selected operations""" | |
| logger.info(f"Starting file processing with operations: {operations}") | |
| self.output_files = [] # Reset output files | |
| results = {} | |
| result_preview = None | |
| if not excel_file: | |
| logger.warning("No file uploaded") | |
| return "Please upload an Excel file", None, None | |
| try: | |
| if any(op in operations for op in ["count_daily_registers_by_source_name", | |
| "count_daily_registers_by_ref", | |
| "count_users_by_source_name", | |
| "count_users_by_ref"]): | |
| try: | |
| df = pd.read_excel(excel_file, sheet_name="User Register") | |
| if "count_daily_registers_by_source_name" in operations: | |
| results["Daily Registers by Source Name"] = self.count_daily_registers_by_source_name(df) | |
| if result_preview is None: | |
| result_preview = results["Daily Registers by Source Name"] | |
| if "count_daily_registers_by_ref" in operations: | |
| results["Daily Registers by Ref"] = self.count_daily_registers_by_ref(df) | |
| if result_preview is None: | |
| result_preview = results["Daily Registers by Ref"] | |
| if "count_users_by_source_name" in operations: | |
| results["Users by Source Name"] = self.count_users_by_source_name(df) | |
| if result_preview is None: | |
| result_preview = results["Users by Source Name"] | |
| if "count_users_by_ref" in operations: | |
| results["Users by Ref"] = self.count_users_by_ref(df) | |
| if result_preview is None: | |
| result_preview = results["Users by Ref"] | |
| except Exception as e: | |
| logger.error(f"Error processing User Register sheet: {str(e)}", exc_info=True) | |
| return f"Error processing User Register sheet: {str(e)}", None, None | |
| if "count_users_each_sheet_by_source_name" in operations: | |
| status, pivot = self.count_users_each_sheet_by_source_name(excel_file) | |
| if status != "Success": | |
| return status, None, None | |
| results["Users Each Sheet by Source Name"] = pivot | |
| if result_preview is None: | |
| result_preview = pivot | |
| if "count_users_each_sheet_by_ref" in operations: | |
| status, pivot = self.count_users_each_sheet_by_ref(excel_file) | |
| if status != "Success": | |
| return status, None, None | |
| results["Users Each Sheet by Ref"] = pivot | |
| if result_preview is None: | |
| result_preview = pivot | |
| if "count_users_each_sheet_by_date" in operations: | |
| status, pivot = self.count_users_each_sheet_by_date(excel_file) | |
| if status != "Success": | |
| return status, None, None | |
| results["Users Each Sheet by Date"] = pivot | |
| if result_preview is None: | |
| result_preview = pivot | |
| if self.output_files: | |
| logger.info("Creating ZIP file with all outputs") | |
| zip_buffer = BytesIO() | |
| with zipfile.ZipFile(zip_buffer, 'w') as zip_file: | |
| for file_path in self.output_files: | |
| if os.path.exists(file_path): | |
| zip_file.write(file_path, os.path.basename(file_path)) | |
| zip_buffer.seek(0) | |
| zip_path = os.path.join("output", "excel_reports.zip") | |
| with open(zip_path, "wb") as f: | |
| f.write(zip_buffer.getvalue()) | |
| logger.info(f"ZIP file created at {zip_path}") | |
| if result_preview is not None and result_preview.size > 10000: | |
| result_preview = result_preview.head(100) | |
| return "Processing completed successfully!", result_preview, zip_path | |
| else: | |
| logger.warning("No operations were performed") | |
| return "No operations were performed.", None, None | |
| except Exception as e: | |
| logger.error(f"Error during file processing: {str(e)}", exc_info=True) | |
| return f"Error: {str(e)}", None, None | |
| # Create the processor | |
| processor = ExcelDataProcessor() | |
| # Define the Gradio interface | |
| with gr.Blocks(title="Excel Data Processor") as app: | |
| gr.Markdown("# Excel Data Processing Tool") | |
| gr.Markdown("Upload your Excel file and select the operations to perform.") | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| file_input = gr.File(label="Upload Excel File") | |
| operations = gr.CheckboxGroup( | |
| choices=[ | |
| "count_daily_registers_by_source_name", | |
| "count_daily_registers_by_ref", | |
| "count_users_by_source_name", | |
| "count_users_by_ref", | |
| "count_users_each_sheet_by_source_name", | |
| "count_users_each_sheet_by_ref", | |
| "count_users_each_sheet_by_date" | |
| ], | |
| label="Select Operations", | |
| value=["count_daily_registers_by_source_name"] | |
| ) | |
| process_btn = gr.Button("Process Excel File", variant="primary") | |
| with gr.Column(scale=2): | |
| status_output = gr.Textbox(label="Status") | |
| with gr.Row(): | |
| result_output = gr.Dataframe(label="Preview Results (Limited to avoid UI freezing)") | |
| download_btn = gr.File(label="Download Results (ZIP)") | |
| def show_processing(file, ops): | |
| logger.info(f"Processing started with operations: {ops}") | |
| return "Processing... This may take a moment. Files are being saved even if UI appears frozen.", None, None | |
| def process_excel_file(file, ops): | |
| logger.info(f"Processing excel file with operations: {ops}") | |
| status, preview, zip_file = processor.process_file(file, ops) | |
| logger.info(f"Processing completed with status: {status}") | |
| return status, preview, zip_file | |
| process_btn.click( | |
| fn=show_processing, | |
| inputs=[file_input, operations], | |
| outputs=[status_output, result_output, download_btn], | |
| queue=False | |
| ).then( | |
| fn=process_excel_file, | |
| inputs=[file_input, operations], | |
| outputs=[status_output, result_output, download_btn] | |
| ) | |
| gr.Markdown(""" | |
| ## Instructions | |
| 1. Upload your Excel file using the file uploader | |
| 2. Select one or more operations to perform | |
| 3. Click "Process Excel File" button | |
| 4. View the results in the preview table (limited to prevent UI freezing) | |
| 5. Download the ZIP file containing all generated Excel files | |
| ## Operations Description | |
| - **count_daily_registers_by_source_name**: Count daily registrations by source name (excluding 'direct') | |
| - **count_daily_registers_by_ref**: Count daily registrations by referral (for 'direct' source only) | |
| - **count_users_by_source_name**: Count unique users by source name (excluding 'direct') | |
| - **count_users_by_ref**: Count unique users by referral (for 'direct' source only) | |
| - **count_users_each_sheet_by_source_name**: Count users in each sheet by source name | |
| - **count_users_each_sheet_by_ref**: Count users in each sheet by referral | |
| - **count_users_each_sheet_by_date**: Count users in each sheet by date | |
| ## Notes | |
| - If the UI appears to freeze, don't worry! The processing is still happening in the background. | |
| - All output files are saved in the 'output' folder even if the UI is unresponsive. | |
| - For very large Excel files, only a preview of the results will be shown to prevent UI freezing. | |
| """) | |
| # Launch the app | |
| if __name__ == "__main__": | |
| app.launch(share=True) |