Update app.py

zip folder download

app.py

@@ -13,7 +13,7 @@ NTU_BLUE = "#003D7C"
 NTU_RED = "#C11E38"
 NTU_GOLD = "#E7B820"
 
-def process_data(file: gr.File, progress=gr.Progress()) -> Tuple[str, str, pd.DataFrame, Union[Tuple[io.BytesIO, str], None]]:
+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:
@@ -29,14 +29,8 @@ def process_data(file: gr.File, progress=gr.Progress()) -> Tuple[str, str, pd.Da
         except Exception as e:
             raise ValueError(f"Error reading Excel file: {str(e)}")
 
-        # Check if required columns are present
-        required_columns = ['user_id', 'lastname', 'course_id']
-        missing_columns = [col for col in required_columns if col not in raw_data.columns]
-        if missing_columns:
-            raise ValueError(f"Missing required columns: {', '.join(missing_columns)}")
-
-        # Extract filename without extension
-        base_filename = os.path.splitext(os.path.basename(file.name))[0]
+        base_path = tempfile.mkdtemp()
+        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()
@@ -60,9 +54,9 @@ def process_data(file: gr.File, progress=gr.Progress()) -> Tuple[str, str, pd.Da
 
         progress(0.6, desc="Calculating grand totals")
 
-        # Step 4: Generate Filenames and Paths (for reference only, not creating actual files)
+        # Step 4: Generate Filenames and Paths
         user_info['File'] = 'User_' + user_info['Username'] + '_data.csv'
-        user_info['Path'] = 'mailmerge/' + user_info['File']
+        user_info['Path'] = user_info['File'].apply(lambda x: os.path.join(base_path, 'mailmerge', x))
 
         # Remove extra columns and summary rows
         user_info = user_info[['Username', 'Name', 'Courses', 'Grand Total', 'Email', 'File', 'Path']]
@@ -72,33 +66,53 @@ def process_data(file: gr.File, progress=gr.Progress()) -> Tuple[str, str, pd.Da
 
         progress(0.8, desc="Generating output files")
 
-        # Create a BytesIO object to store the zip file
-        zip_buffer = io.BytesIO()
+        # Calculate totals for Courses and Grand Total
+        total_courses = user_info['Courses'].sum()
+        total_grand_total = user_info['Grand Total'].sum()
 
-        with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file:
-            # Save individual CSV files
-            for user_id in user_info['Username'].unique():
-                user_data = raw_data[raw_data['user_id'] == user_id][required_columns]
-                user_file_path = f'mailmerge/User_{user_id}_data.csv'
-                zip_file.writestr(user_file_path, user_data.to_csv(index=False))
-
-            # Save the final Excel file
-            excel_buffer = io.BytesIO()
-            with pd.ExcelWriter(excel_buffer, engine='xlsxwriter') as writer:
-                user_info.to_excel(writer, index=False, sheet_name='Sheet1')
-                workbook = writer.book
-                worksheet = writer.sheets['Sheet1']
-                
-                last_row = len(user_info) + 1
-                worksheet.write(f'B{last_row + 1}', 'Total')
-                worksheet.write(f'C{last_row + 1}', user_info['Courses'].sum())
-                worksheet.write(f'D{last_row + 1}', 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']
+        mailmerge_path = os.path.join(base_path, 'mailmerge')
+        if not os.path.exists(mailmerge_path):
+            os.makedirs(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(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']
             
-            zip_file.writestr(f'mailmerge {base_filename}.xlsx', excel_buffer.getvalue())
+            # 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(mailmerge_path):
+                for file in files:
+                    file_path = os.path.join(root, file)
+                    zip_file.write(file_path, os.path.relpath(file_path, base_path))
+
+            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, (zip_buffer, f"gradebook_results_{base_filename}.zip")
+        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
@@ -148,7 +162,7 @@ def create_scatter_plot(df: pd.DataFrame) -> Union[px.scatter, None]:
         print(f"Error creating scatter plot: {str(e)}")
         return None
 
-def update_insights(df: pd.DataFrame, zip_data: Union[Tuple[io.BytesIO, str], None]) -> List[Union[gr.components.Component, 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
@@ -162,9 +176,8 @@ def update_insights(df: pd.DataFrame, zip_data: Union[Tuple[io.BytesIO, str], No
         
         user_table = gr.DataFrame(value=df)
         
-        if zip_data:
-            zip_file, zip_name = zip_data
-            download_button = gr.File(value=zip_file, filename=zip_name, visible=True, label="Download Results")
+        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")
@@ -177,12 +190,12 @@ def update_insights(df: pd.DataFrame, zip_data: Union[Tuple[io.BytesIO, str], No
 
 def process_and_update(file):
     try:
-        result_msg, csv_loc, df, zip_data = process_data(file)
-        insights = update_insights(df, zip_data)
-        return [result_msg, csv_loc] + insights
+        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"] + [gr.Markdown(error_msg)] + [None] * 6
+        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
@@ -201,40 +214,7 @@ custom_theme = gr.themes.Base().set(
     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:
+with gr.Blocks(theme=custom_theme, css="custom.css") as iface:
     gr.Markdown("# Gradebook Data Processor", elem_classes=["app-title"])
     
     with gr.Tabs():
@@ -244,6 +224,7 @@ with gr.Blocks(theme=custom_theme, css=custom_css) as iface:
             process_btn = gr.Button("Process Data", variant="primary")
             output_msg = gr.Textbox(label="Processing Result")
             csv_location = gr.Textbox(label="Output Information")
+            download_button = gr.File(visible=False, label="Download Results")
             gr.Markdown("After processing, switch to the 'Data Insights' tab to view results and download files.")
         
         with gr.TabItem("2. Data Insights Dashboard"):
@@ -256,10 +237,6 @@ with gr.Blocks(theme=custom_theme, css=custom_css) as iface:
             
             scatter_plot = gr.Plot()
             user_table = gr.DataFrame()
-            
-            gr.Markdown("## Download Processed Data")
-            download_button = gr.File(visible=False, label="Download Results")
-            download_text = gr.Markdown("")
     
     clear_btn = gr.Button("Clear All Data", variant="secondary")
     gr.Markdown("Click 'Clear All Data' to reset the application and start over.")
@@ -267,14 +244,14 @@ with gr.Blocks(theme=custom_theme, css=custom_css) as iface:
     process_btn.click(
         process_and_update,
         inputs=[file_input],
-        outputs=[output_msg, csv_location, summary_stats, users_activity_chart, users_courses_chart, scatter_plot, user_table, download_button, download_text]
+        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, download_text]
+        outputs=[output_msg, csv_location, summary_stats, users_activity_chart, users_courses_chart, scatter_plot, user_table, download_button]
     )
 
 if __name__ == "__main__":
-    iface.launch()
+    iface.launch()