import pandas as pd import os import gradio as gr import plotly.express as px from typing import Tuple, List, Union import traceback import io import zipfile import tempfile # NTU Singapore colors NTU_BLUE = "#003D7C" NTU_RED = "#C11E38" NTU_GOLD = "#E7B820" def process_data(file: gr.File, progress=gr.Progress()) -> Tuple[str, str, pd.DataFrame, Union[str, None]]: try: # Check if file is uploaded if file is None: raise ValueError("No file uploaded. Please upload an Excel file.") # Check file extension if not file.name.lower().endswith(('.xls', '.xlsx')): raise ValueError("Invalid file format. Please upload an Excel file (.xls or .xlsx).") # Load the raw Excel file try: raw_data = pd.read_excel(file.name) except Exception as e: raise ValueError(f"Error reading Excel file: {str(e)}") base_path = tempfile.mkdtemp() mailmerge_path = os.path.join('mailmerge') final_file_path = os.path.join(base_path, 'final_output.xlsx') # Step 1: Extract User Information user_info = raw_data[['user_id', 'lastname']].drop_duplicates().copy() user_info['Username'] = user_info['user_id'] user_info['Name'] = user_info['lastname'] user_info['Email'] = user_info['user_id'] + '@ntu.edu.sg' progress(0.2, desc="Extracting user information") # Step 2: Calculate Course Count course_counts = raw_data.groupby('user_id')['course_id'].nunique().reset_index() course_counts.columns = ['Username', 'Courses'] user_info = user_info.merge(course_counts, on='Username', how='left') progress(0.4, desc="Calculating course counts") # Step 3: Calculate Grand Total event_counts = raw_data.groupby('user_id').size().reset_index(name='Grand Total') event_counts.columns = ['Username', 'Grand Total'] user_info = user_info.merge(event_counts, on='Username', how='left') progress(0.6, desc="Calculating grand totals") # Step 4: Generate Filenames and Paths user_info['File'] = 'User_' + user_info['Username'] + '_data.csv' user_info['Path'] = user_info['File'].apply(lambda x: os.path.join(mailmerge_path, x)) # Remove extra columns and summary rows user_info = user_info[['Username', 'Name', 'Courses', 'Grand Total', 'Email', 'File', 'Path']] user_info = user_info[user_info['Username'].notna()] user_info.drop_duplicates(subset=['Username'], inplace=True) user_info.sort_values(by='Username', inplace=True) progress(0.8, desc="Generating output files") # Calculate totals for Courses and Grand Total total_courses = user_info['Courses'].sum() total_grand_total = user_info['Grand Total'].sum() # Generate individual CSV files for each user required_columns = ['course_id', 'course_pk1', 'data', 'event_type', 'internal_handle', 'lastname', 'session_id', 'timestamp', 'user_id', 'system_role'] full_mailmerge_path = os.path.join(base_path, mailmerge_path) if not os.path.exists(full_mailmerge_path): os.makedirs(full_mailmerge_path) for user_id in user_info['Username'].unique(): user_data = raw_data[raw_data['user_id'] == user_id][required_columns] user_file_path = os.path.join(full_mailmerge_path, f'User_{user_id}_data.csv') user_data.to_csv(user_file_path, index=False) # Save the final dataframe to the output Excel file with pd.ExcelWriter(final_file_path, engine='xlsxwriter') as writer: user_info.to_excel(writer, index=False, sheet_name='Sheet1') workbook = writer.book worksheet = writer.sheets['Sheet1'] # Find the last row number dynamically last_row = len(user_info) + 1 # Account for header row in Excel # Write the total values in columns B, C, and D of the first empty row after the user data worksheet.write(f'B{last_row + 1}', 'Total') worksheet.write(f'C{last_row + 1}', total_courses) worksheet.write(f'D{last_row + 1}', total_grand_total) # Create a zip file containing all user CSV files and the final Excel file zip_buffer = io.BytesIO() with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file: for root, _, files in os.walk(full_mailmerge_path): for file in files: file_path = os.path.join(root, file) arcname = os.path.relpath(file_path, base_path) zip_file.write(file_path, arcname) zip_file.write(final_file_path, os.path.basename(final_file_path)) zip_buffer.seek(0) temp_zip_file = tempfile.NamedTemporaryFile(delete=False, suffix=".zip") with open(temp_zip_file.name, 'wb') as f: f.write(zip_buffer.getvalue()) progress(1.0, desc="Processing complete") return "Processing complete. You can now download the results.", "Results are packaged in the zip file below.", user_info, temp_zip_file.name except Exception as e: error_msg = f"Error: {str(e)}\n\nTraceback:\n{traceback.format_exc()}" return error_msg, "Processing failed", pd.DataFrame(), None def create_summary_stats(df: pd.DataFrame) -> dict: try: return { "Total Users": len(df), "Total Courses": df['Courses'].sum(), "Total Activity": df['Grand Total'].sum(), "Avg Courses per User": df['Courses'].mean(), "Avg Activity per User": df['Grand Total'].mean() } except Exception as e: return {"Error": f"Failed to create summary stats: {str(e)}"} def create_bar_chart(df: pd.DataFrame, x: str, y: str, title: str) -> Union[px.bar, None]: try: if df.empty: return None fig = px.bar(df, x=x, y=y, title=title) fig.update_layout( plot_bgcolor='white', paper_bgcolor='white', font_color=NTU_BLUE ) fig.update_xaxes(tickmode='linear') # Ensure all x-axis labels are shown fig.update_traces(marker_color=NTU_BLUE) return fig except Exception as e: print(f"Error creating bar chart: {str(e)}") return None def create_scatter_plot(df: pd.DataFrame) -> Union[px.scatter, None]: try: if df.empty: return None fig = px.scatter(df, x='Courses', y='Grand Total', title='Courses vs. Activity Level', hover_data=['Username', 'Name']) fig.update_layout( plot_bgcolor='white', paper_bgcolor='white', font_color=NTU_BLUE ) fig.update_xaxes(tickmode='linear') # Ensure all x-axis labels are shown fig.update_traces(marker_color=NTU_RED) return fig except Exception as e: print(f"Error creating scatter plot: {str(e)}") return None def update_insights(df: pd.DataFrame, zip_path: Union[str, None]) -> List[Union[gr.components.Component, None]]: try: if df.empty: return [gr.Markdown("No data available. Please upload and process a file first.")] + [None] * 6 stats = create_summary_stats(df) stats_md = gr.Markdown("\n".join([f"**{k}**: {v:.2f}" for k, v in stats.items()])) users_activity_chart = create_bar_chart(df, 'Username', 'Grand Total', 'User Activity Levels') users_courses_chart = create_bar_chart(df, 'Username', 'Courses', 'Courses per User') scatter_plot = create_scatter_plot(df) user_table = gr.DataFrame(value=df) if zip_path: download_button = gr.File(value=zip_path, visible=True, label="Download Results") download_text = gr.Markdown("Click the 'Download Results' button above to download the ZIP file containing all processed data.") else: download_button = gr.File(visible=False, label="Download Results") download_text = gr.Markdown("") return [stats_md, users_activity_chart, users_courses_chart, scatter_plot, user_table, download_button, download_text] except Exception as e: error_msg = f"Error updating insights: {str(e)}\n\nTraceback:\n{traceback.format_exc()}" return [gr.Markdown(error_msg)] + [None] * 6 def process_and_update(file): try: result_msg, csv_loc, df, zip_path = process_data(file) insights = update_insights(df, zip_path) return [result_msg, csv_loc, zip_path] + insights except Exception as e: error_msg = f"Error in process_and_update: {str(e)}\n\nTraceback:\n{traceback.format_exc()}" return [error_msg, "Processing failed", None] + [gr.Markdown(error_msg)] + [None] * 6 def clear_outputs(): return [""] * 2 + [None] * 6 + [""] # 2 text outputs, 6 graph/table/file outputs, and 1 download text # Create a custom theme custom_theme = gr.themes.Base().set( body_background_fill="#E6F3FF", body_text_color="#003D7C", button_primary_background_fill="#C11E38", button_primary_background_fill_hover="#A5192F", button_primary_text_color="white", block_title_text_color="#003D7C", block_label_background_fill="#E6F3FF", input_background_fill="white", input_border_color="#003D7C", input_border_color_focus="#C11E38", ) # Custom CSS custom_css = """ .gr-button-secondary { background-color: #F0F0F0; color: #003D7C; border: 1px solid #003D7C; border-radius: 12px; padding: 8px 16px; font-size: 16px; font-weight: bold; cursor: pointer; transition: background-color 0.3s, color 0.3s, border-color 0.3s; } .gr-button-secondary:hover { background-color: #003D7C; color: white; border-color: #003D7C; } .gr-button-secondary:active { transform: translateY(1px); } .app-title { color: #003D7C; font-size: 24px; font-weight: bold; text-align: center; margin-bottom: 20px; } """ with gr.Blocks(theme=custom_theme, css=custom_css) as iface: gr.Markdown("# Gradebook Data Processor", elem_classes=["app-title"]) with gr.Tabs(): with gr.TabItem("1. File Upload and Processing"): gr.Markdown("## Step 1: Upload your Excel file and process the data") file_input = gr.File(label="Upload Excel File") process_btn = gr.Button("Process Data", variant="primary") output_msg = gr.Textbox(label="Processing Result") csv_location = gr.Textbox(label="Output Information") gr.Markdown("## Step 2: Download Mail merge package") download_button = gr.File(visible=False, label="Download Results") gr.Markdown("Tip: 😎 After processing, switch to the 'Data Insights' tab at the top to view results and analytics.") with gr.TabItem("2. Data Insights Dashboard"): gr.Markdown("## Data Insights") summary_stats = gr.Markdown("Upload and process a file to see summary statistics.") with gr.Row(): users_activity_chart = gr.Plot() users_courses_chart = gr.Plot() scatter_plot = gr.Plot() user_table = gr.DataFrame() clear_btn = gr.Button("Clear All Data", variant="secondary") gr.Markdown("Click 'Clear All Data' to reset the application and start over.") process_btn.click( process_and_update, inputs=[file_input], outputs=[output_msg, csv_location, download_button, summary_stats, users_activity_chart, users_courses_chart, scatter_plot, user_table, download_button] ) clear_btn.click( clear_outputs, inputs=[], outputs=[output_msg, csv_location, summary_stats, users_activity_chart, users_courses_chart, scatter_plot, user_table, download_button] ) if __name__ == "__main__": iface.launch()